diff --git a/CHANGELOG.md b/CHANGELOG.md
index 96be8fab..87bdfe4d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -81,6 +81,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
Compatibility note: before upgrading, check that applications do not send these writes through Proxy while encrypted columns are configured: `MERGE`; SQL `PREPARE` of a write; a write inside `COPY (...) TO`; bulk `UPDATE ... FROM (VALUES ...)`; `INSERT ... SELECT ... FROM unnest(...)`; and other writes Proxy cannot type check, including writes to tables with no encrypted columns. Reads that use an operation an encrypted column does not support, including `ORDER BY`, `GROUP BY`, `DISTINCT` or `UNION` over a column without the index they need, are now refused too.
+- **`COPY ... FROM` into encrypted columns**: when encrypted columns are configured, Proxy now refuses a `COPY ... FROM` (from the client, a file or a program) that would fill an encrypted column, with an error that names the column, as Proxy does not encrypt `COPY` data. Previously such a `COPY` was forwarded, and stored plaintext in the encrypted column. A `COPY` with no column list fills every column, so it is refused when the table has any encrypted column. A `COPY ... FROM` into a table Proxy cannot resolve, or into a column it has not loaded, is refused too, whatever `CS_DEVELOPMENT__ENABLE_MAPPING_ERRORS` is set to. A simple query that has a `COPY` and a statement Proxy encrypts is refused; send the `COPY` in a query of its own. See [COPY into encrypted column](docs/errors.md#mapping-copy-into-encrypted-column).
+
+- **Statements after `COPY ... FROM STDIN`**: Proxy now forwards a `COPY ... FROM STDIN` into columns that are not encrypted, so `psql`'s `\copy` and driver `COPY` APIs can bulk-load plaintext tables through Proxy. Previously Proxy refused the statement as invalid SQL unless it ended with `;`. With a `;`, Proxy read every later statement in the same simple query as copy data and forwarded it unchecked, so a plaintext `INSERT` into an encrypted column could reach PostgreSQL. Those statements are now checked and encrypted like any other.
+
+ Compatibility note: before upgrading, check that applications do not `COPY` into encrypted columns through Proxy; load those columns with `INSERT` instead.
+
- **Short encryption results**: Proxy now refuses a statement when the encryption service returns fewer values than it was given, instead of sending the unmatched parameters unencrypted. See [Encryption result count mismatch](docs/errors.md#encrypt-result-count-mismatch).
- **Updated PostgreSQL client library**: Proxy is now built against `tokio-postgres` 0.7.18, which resolves [GHSA-3gjw-f78c-vvpw](https://github.com/advisories/GHSA-3gjw-f78c-vvpw). No configuration change is required.
diff --git a/Cargo.lock b/Cargo.lock
index 1cd82195..b1cbaf2f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -910,6 +910,7 @@ dependencies = [
"cipherstash-config",
"cipherstash-proxy",
"fake 4.2.0",
+ "futures-util",
"hex",
"postgres-types",
"rand 0.9.3",
diff --git a/docs/errors.md b/docs/errors.md
index c2379856..0bb5be90 100644
--- a/docs/errors.md
+++ b/docs/errors.md
@@ -19,6 +19,9 @@
- [Unmappable encrypted column](#mapping-unmappable-encrypted-column)
- [Unresolvable table](#mapping-unresolvable-table)
- [Ambiguous table](#mapping-ambiguous-table)
+ - [COPY into encrypted column](#mapping-copy-into-encrypted-column)
+ - [COPY into unknown column](#mapping-copy-into-unknown-column)
+ - [COPY with a statement that uses an encrypted column](#mapping-copy-with-encrypted-statement)
- [Unsupported encrypted operation](#mapping-unsupported-encrypted-operation)
- [Unknown prepared statement](#mapping-unknown-prepared-statement)
- [Internal Error](#mapping-internal-error)
@@ -488,6 +491,80 @@ It does not track `SET search_path`, so a client that puts `pg_temp` later in it
+## COPY into encrypted column
+
+A `COPY ... FROM` statement would load data into an encrypted column.
+CipherStash Proxy does not encrypt `COPY` data, so it refuses the statement rather than store plaintext in the encrypted column.
+
+### Error message
+
+```
+COPY into encrypted column 'users.encrypted_email' is not supported, as Proxy does not encrypt COPY data.
+```
+
+### Notes
+
+The refusal applies to a `COPY ... FROM STDIN`, and to a `COPY ... FROM` a file or a program.
+A `COPY` with no column list loads every column of the table, so it is refused when the table has any encrypted column.
+A `COPY ... FROM` into a table with no encrypted columns, or into only the plaintext columns of a table, is forwarded to PostgreSQL with its data.
+A `COPY ... TO` of a table or of a query that reads is forwarded, and returns encrypted columns as ciphertext, as Proxy does not decrypt `COPY` data.
+A `COPY ... FROM` into a table Proxy cannot resolve is refused as an [Unresolvable table](#mapping-unresolvable-table) or an [Ambiguous table](#mapping-ambiguous-table).
+That includes a table created with `CREATE TABLE ... (LIKE ...)` or `CREATE TABLE ... AS`, whose columns Proxy does not know.
+When no columns are encrypted, Proxy does not check the table of a `COPY`.
+A `COPY` after DDL that may change which columns are encrypted, in the same simple query, is still refused, as an `INSERT` would be.
+
+### How to fix
+
+Load the encrypted columns with `INSERT` statements through CipherStash Proxy, which encrypts the values.
+Use `COPY` only for columns that are not encrypted.
+
+
+
+
+
+## COPY into unknown column
+
+A `COPY ... FROM` statement names a column that is not in the database schema CipherStash Proxy has loaded for its table.
+Proxy cannot tell whether the column is encrypted, so it refuses the statement.
+
+### Error message
+
+```
+COPY into column 'users.nickname', which is not in the database schema Proxy has loaded, is not supported, as Proxy cannot tell whether it is encrypted.
+```
+
+### Notes
+
+The refusal applies only when any column is encrypted, and whatever `CS_DEVELOPMENT__ENABLE_MAPPING_ERRORS` is set to.
+
+### How to fix
+
+Check that the column exists and that its name is spelled as PostgreSQL resolves it.
+If the column was added on another connection, retry once Proxy has reloaded the database schema.
+
+
+
+
+
+## COPY with a statement that uses an encrypted column
+
+A simple query contains a `COPY` together with a statement that uses an encrypted column.
+CipherStash Proxy rewrites such a query to encrypt it, and cannot keep the `COPY` in the rewritten query.
+
+### Error message
+
+```
+COPY must be sent in a query of its own when the query has a statement that uses an encrypted column.
+```
+
+### How to fix
+
+Send the `COPY` in a query of its own.
+
+
+
+
+
## Unsupported encrypted operation
A statement uses an operation on an encrypted column that the column's EQL type does not support, such as `<` or `ORDER BY` on a column with no ordering index, or `LIKE` on a column with no match index.
diff --git a/packages/cipherstash-proxy-integration/Cargo.toml b/packages/cipherstash-proxy-integration/Cargo.toml
index b1a011cc..11813d0b 100644
--- a/packages/cipherstash-proxy-integration/Cargo.toml
+++ b/packages/cipherstash-proxy-integration/Cargo.toml
@@ -16,6 +16,7 @@ cipherstash-config = { workspace = true }
cipherstash-proxy = { path = "../cipherstash-proxy/" }
chrono = { version = "0.4.39", features = ["clock"] }
fake = { version = "4", features = ["chrono", "derive"] }
+futures-util = { version = "0.3", features = ["sink"] }
hex = "0.4.3"
postgres-types = { version = "0.2.9", features = ["derive"] }
rand = "0.9"
diff --git a/packages/cipherstash-proxy-integration/src/copy_from_stdin.rs b/packages/cipherstash-proxy-integration/src/copy_from_stdin.rs
new file mode 100644
index 00000000..d1fa6d29
--- /dev/null
+++ b/packages/cipherstash-proxy-integration/src/copy_from_stdin.rs
@@ -0,0 +1,68 @@
+//! `COPY ... FROM STDIN` through Proxy, which tokio-postgres sends with the extended protocol.
+//! Proxy forwards a copy into a table with no encrypted columns, and refuses one into an
+//! encrypted column, as it does not encrypt copy data.
+
+#[cfg(test)]
+mod tests {
+ use crate::common::{clear, connect_with_tls, random_id, trace, PROXY};
+ use bytes::Bytes;
+ use futures_util::SinkExt;
+ use std::pin::pin;
+
+ #[tokio::test]
+ async fn copy_into_a_plaintext_table() {
+ trace();
+ clear().await;
+ let client = connect_with_tls(*PROXY).await;
+
+ let (first, second) = (random_id(), random_id());
+ let mut sink = pin!(client
+ .copy_in("COPY plaintext (id, plaintext) FROM STDIN")
+ .await
+ .unwrap());
+ sink.send(Bytes::from(format!("{first}\talice\n{second}\tbob\n")))
+ .await
+ .unwrap();
+ assert_eq!(sink.as_mut().finish().await.unwrap(), 2);
+
+ let rows = client
+ .query(
+ "SELECT plaintext FROM plaintext WHERE id = ANY($1) ORDER BY plaintext",
+ &[&vec![first, second]],
+ )
+ .await
+ .unwrap();
+ let copied: Vec = rows.iter().map(|row| row.get(0)).collect();
+ assert_eq!(copied, vec!["alice", "bob"]);
+ }
+
+ #[tokio::test]
+ async fn copy_into_an_encrypted_column_is_refused() {
+ trace();
+ clear().await;
+ let client = connect_with_tls(*PROXY).await;
+
+ // tokio-postgres' `Display` no longer includes the server's message
+ let refusal = |err: tokio_postgres::Error| err.as_db_error().unwrap().message().to_owned();
+ let expected = "COPY into encrypted column 'encrypted.encrypted_text' is not supported";
+
+ let err = client
+ .copy_in::<_, Bytes>("COPY encrypted (id, encrypted_text) FROM STDIN")
+ .await
+ .err()
+ .unwrap();
+ assert!(refusal(err).starts_with(expected));
+
+ // A COPY with no column list fills every column, including the encrypted ones
+ let err = client
+ .copy_in::<_, Bytes>("COPY encrypted FROM STDIN")
+ .await
+ .err()
+ .unwrap();
+ assert!(refusal(err).starts_with(expected));
+
+ // The connection is still usable
+ let rows = client.simple_query("SELECT 1").await.unwrap();
+ assert!(!rows.is_empty());
+ }
+}
diff --git a/packages/cipherstash-proxy-integration/src/lib.rs b/packages/cipherstash-proxy-integration/src/lib.rs
index f88505f5..ad4b3493 100644
--- a/packages/cipherstash-proxy-integration/src/lib.rs
+++ b/packages/cipherstash-proxy-integration/src/lib.rs
@@ -1,5 +1,6 @@
mod common;
mod connection_resilience;
+mod copy_from_stdin;
mod decrypt;
mod diagnostics;
mod disable_mapping;
diff --git a/packages/cipherstash-proxy/CONTEXT.md b/packages/cipherstash-proxy/CONTEXT.md
index 41d33c03..5ee4d621 100644
--- a/packages/cipherstash-proxy/CONTEXT.md
+++ b/packages/cipherstash-proxy/CONTEXT.md
@@ -67,6 +67,7 @@ reason, or refuse it. Refusing is the default. A failed type check falls back to
forwarding only for a coverage gap (the statement could not be type checked, or an
internal mapper error), only while mapping errors are off, and never in a simple Query with another statement to map.
When encrypted columns are configured, a write never falls back.
+A `COPY ... FROM` is not type checked: when encrypted columns are configured, it is forwarded only if every column it fills resolves to a native column.
Each statement in a simple Query is type checked against a **batch view**: the effective schema plus the encryption-neutral DDL of the statements before it.
**Connection temporary tables**:
diff --git a/packages/cipherstash-proxy/src/error.rs b/packages/cipherstash-proxy/src/error.rs
index 4921ab72..3fa962fe 100644
--- a/packages/cipherstash-proxy/src/error.rs
+++ b/packages/cipherstash-proxy/src/error.rs
@@ -148,6 +148,15 @@ pub enum MappingError {
#[error("Statement refers to table '{table}', which names both a temporary table and a permanent table, so Proxy cannot tell which one it writes to. Qualify the name, as in 'pg_temp.{table}'. For help visit {}#mapping-ambiguous-table", ERROR_DOC_BASE_URL)]
AmbiguousTable { table: String },
+ #[error("COPY into encrypted column '{table}.{column}' is not supported, as Proxy does not encrypt COPY data. For help visit {}#mapping-copy-into-encrypted-column", ERROR_DOC_BASE_URL)]
+ CopyIntoEncryptedColumn { table: String, column: String },
+
+ #[error("COPY into column '{table}.{column}', which is not in the database schema Proxy has loaded, is not supported, as Proxy cannot tell whether it is encrypted. For help visit {}#mapping-copy-into-unknown-column", ERROR_DOC_BASE_URL)]
+ CopyIntoUnknownColumn { table: String, column: String },
+
+ #[error("COPY must be sent in a query of its own when the query has a statement that uses an encrypted column. For help visit {}#mapping-copy-with-encrypted-statement", ERROR_DOC_BASE_URL)]
+ CopyWithEncryptedStatement,
+
#[error("Statement uses an operation that an encrypted column's EQL type does not support: {}. For help visit {}#mapping-unsupported-encrypted-operation", _0, ERROR_DOC_BASE_URL)]
UnsupportedEncryptedOperation(String),
diff --git a/packages/cipherstash-proxy/src/postgresql/context/lifecycle.rs b/packages/cipherstash-proxy/src/postgresql/context/lifecycle.rs
index ec3c9f3c..5b399ecf 100644
--- a/packages/cipherstash-proxy/src/postgresql/context/lifecycle.rs
+++ b/packages/cipherstash-proxy/src/postgresql/context/lifecycle.rs
@@ -146,7 +146,7 @@ impl Context {
}
};
self.prepare_schema_for_statement().await?;
- if eql_mapper::requires_type_check(statement) {
+ if eql_mapper::requires_schema(statement) {
self.wait_for_schema_execution().await;
self.ensure_schema_modelled()?;
}
diff --git a/packages/cipherstash-proxy/src/postgresql/context/mod.rs b/packages/cipherstash-proxy/src/postgresql/context/mod.rs
index 6dd09870..44077828 100644
--- a/packages/cipherstash-proxy/src/postgresql/context/mod.rs
+++ b/packages/cipherstash-proxy/src/postgresql/context/mod.rs
@@ -1539,7 +1539,7 @@ where
for (statement, _, changes) in
settings_protocol::project_settings_effects(&state.settings, statements)?
{
- if changed && eql_mapper::requires_type_check(statement) {
+ if changed && eql_mapper::requires_schema(statement) {
return Ok(true);
}
changed |= changes;
diff --git a/packages/cipherstash-proxy/src/postgresql/context/settings_protocol.rs b/packages/cipherstash-proxy/src/postgresql/context/settings_protocol.rs
index 06f01d8b..0091d56f 100644
--- a/packages/cipherstash-proxy/src/postgresql/context/settings_protocol.rs
+++ b/packages/cipherstash-proxy/src/postgresql/context/settings_protocol.rs
@@ -170,7 +170,7 @@ impl SettingsProtocol {
Prepared {
operation,
command,
- needs_mapping: eql_mapper::requires_type_check(statement),
+ needs_mapping: eql_mapper::requires_schema(statement),
settings,
literal_sensitive: false,
},
diff --git a/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs b/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs
index f604358b..1c28d041 100644
--- a/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs
+++ b/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs
@@ -28,8 +28,8 @@ use crate::proxy::EncryptionService;
use crate::{EqlOutput, EqlQueryPayload};
use cipherstash_client::encryption::Plaintext;
use eql_mapper::{
- self, EqlMapperError, EqlTermVariant, JsonSelectorSegment, TableResolver, TypeCheckedStatement,
- UnresolvedTable,
+ self, EqlMapperError, EqlTermVariant, JsonSelectorSegment, SchemaError, TableResolver,
+ TypeCheckedStatement, UnresolvedTable,
};
use metrics::{counter, histogram};
use pg_proto::{Describe, Execute, FrontendMessage, FrontendMiddlewareOutput, Parse};
@@ -432,10 +432,7 @@ impl Frontend {
{
return Err(MappingError::DependentStatementAfterDdl.into());
}
- if parsed_statements
- .iter()
- .any(eql_mapper::requires_type_check)
- {
+ if parsed_statements.iter().any(eql_mapper::requires_schema) {
self.context.wait_for_schema_execution().await;
self.context.ensure_schema_modelled()?;
}
@@ -454,6 +451,16 @@ impl Frontend {
.map(|(statement, resolver)| self.route(statement, vec![], &resolver))
.collect::, _>>()?;
+ // Proxy writes out every statement again when it rewrites a Query, and
+ // a `COPY` written out again is not the one the client sent.
+ if routes.iter().any(Route::is_map)
+ && parsed_statements
+ .iter()
+ .any(|statement| matches!(statement, ast::Statement::Copy { .. }))
+ {
+ return Err(MappingError::CopyWithEncryptedStatement.into());
+ }
+
// A fallback forwards the whole Query as the client wrote it, so it
// must not carry a statement that needs mapping: that statement's
// values would reach the database unencrypted.
@@ -1247,6 +1254,12 @@ impl Frontend {
if self.context.unsafe_disable_mapping()? {
return Ok(Route::Forward(Forward::MappingDisabled));
}
+ if let Some((table_name, columns)) = eql_mapper::copies_into_table(statement) {
+ if self.context.has_encrypted_columns() {
+ check_copy_into(table_name, columns, resolver)?;
+ }
+ return Ok(Route::Forward(Forward::NothingEncrypted));
+ }
if !eql_mapper::requires_type_check(statement) {
// A write nested in a statement the type checker skips, such as
// `COPY (INSERT ... RETURNING ...) TO STDOUT`, is never mapped.
@@ -1380,6 +1393,67 @@ impl Frontend {
}
}
+/// Refuses a `COPY ... FROM` into `columns` of `table_name`, or every column
+/// when `columns` is empty, unless `resolver` shows none of them is encrypted.
+/// Proxy does not encrypt copy data, and cannot tell whether a table or column
+/// it has not loaded is encrypted.
+fn check_copy_into(
+ table_name: &ast::ObjectName,
+ columns: &[ast::Ident],
+ resolver: &TableResolver,
+) -> Result<(), Error> {
+ let table = resolver.resolve_table(table_name).map_err(|err| {
+ let table = table_name.to_string();
+ match err {
+ SchemaError::AmbiguousTable(_) => MappingError::AmbiguousTable { table },
+ _ => MappingError::UnresolvableTable { table },
+ }
+ })?;
+ let unknown = |column: &ast::Ident| MappingError::CopyIntoUnknownColumn {
+ table: table.name.value.clone(),
+ column: column.value.clone(),
+ };
+
+ let targets = if columns.is_empty() {
+ table.columns.clone()
+ } else {
+ columns
+ .iter()
+ .map(|column| table.get_column(column).map_err(|_| unknown(column)))
+ .collect::>()?
+ };
+ // A defence: a table with no columns Proxy knows. A loaded table has its
+ // columns, and DDL that leaves a table's columns unknown, such as
+ // `CREATE TABLE ... AS`, fails closed before a `COPY` in the same Query.
+ if targets.is_empty() {
+ return Err(MappingError::UnresolvableTable {
+ table: table_name.to_string(),
+ }
+ .into());
+ }
+
+ for column in targets {
+ // The parser reads the `LIKE` clause of `CREATE TABLE t (LIKE u)` as a
+ // column named `LIKE`, so Proxy does not know the columns it copies.
+ // PostgreSQL reserves `LIKE`, so an unquoted column of that name is
+ // always the clause.
+ if column.name.quote_style.is_none() && column.name.value.eq_ignore_ascii_case("like") {
+ return Err(MappingError::UnresolvableTable {
+ table: table_name.to_string(),
+ }
+ .into());
+ }
+ if !matches!(column.kind, eql_mapper::ColumnKind::Native) {
+ return Err(MappingError::CopyIntoEncryptedColumn {
+ table: table.name.value.clone(),
+ column: column.name.value.clone(),
+ }
+ .into());
+ }
+ }
+ Ok(())
+}
+
/// The prepared statements a `DEALLOCATE` or `DISCARD ALL` statement removes, if any
fn to_deallocation(statement: &ast::Statement) -> Option {
match statement {
@@ -4405,4 +4479,179 @@ mod unresolvable_table_tests {
let sql = "SELECT * FROM unknown_view; SELECT id FROM encrypted";
assert_eq!(query(&mut frontend, sql).await, sql);
}
+
+ fn copy_into_encrypted_column(table: &str, column: &str) -> String {
+ MappingError::CopyIntoEncryptedColumn {
+ table: table.to_owned(),
+ column: column.to_owned(),
+ }
+ .to_string()
+ }
+
+ /// Whether `forwarded` is the refusal Proxy sends for `message`, quoted once
+ /// in the DO block's body and again as the DO block's literal.
+ fn is_refusal(forwarded: &str, message: &str) -> bool {
+ forwarded.contains("RAISE EXCEPTION") && forwarded.contains(&message.replace('\'', "''''"))
+ }
+
+ /// Proxy does not encrypt copy data, so a `COPY ... FROM` that fills an
+ /// encrypted column would store plaintext in it, from a client stream, a
+ /// file or a program alike.
+ #[tokio::test]
+ async fn copy_into_an_encrypted_column_is_refused() {
+ for sql in [
+ "COPY encrypted FROM STDIN",
+ "COPY encrypted (secret) FROM STDIN;",
+ "COPY public.encrypted (id, SECRET) FROM STDIN WITH (FORMAT csv)",
+ "COPY encrypted (id, secret) FROM '/tmp/secrets.csv'",
+ "COPY encrypted FROM PROGRAM 'cat /tmp/secrets.csv'",
+ "COPY notes FROM STDIN; COPY encrypted FROM STDIN",
+ ] {
+ let (mut frontend, _) = encrypted_client();
+ let forwarded = query(&mut frontend, sql).await;
+ assert!(
+ is_refusal(
+ &forwarded,
+ ©_into_encrypted_column("encrypted", "secret")
+ ),
+ "{sql}: {forwarded}"
+ );
+ }
+
+ let (mut frontend, _) = encrypted_client();
+ let result = frontend
+ .parse_handler(operation_id(), parse("COPY encrypted FROM STDIN"))
+ .await;
+ assert!(
+ matches!(
+ &result,
+ Err(Error::Mapping(MappingError::CopyIntoEncryptedColumn { table, column }))
+ if table == "encrypted" && column == "secret"
+ ),
+ "{result:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn copy_into_native_columns_is_forwarded() {
+ for sql in [
+ "COPY notes FROM STDIN",
+ "COPY notes (body) FROM STDIN;",
+ "COPY encrypted (id) FROM STDIN WITH (FORMAT csv)",
+ "COPY notes FROM '/tmp/notes.csv'",
+ "COPY encrypted TO STDOUT",
+ "CREATE TEMPORARY TABLE names (name text); COPY names FROM STDIN",
+ "CREATE TABLE names (name text); COPY names (name) FROM STDIN",
+ ] {
+ let (mut frontend, _) = encrypted_client();
+ assert_eq!(query(&mut frontend, sql).await, sql, "{sql}");
+ }
+
+ let (mut frontend, _) = encrypted_client();
+ let result = frontend
+ .parse_handler(operation_id(), parse("COPY notes FROM STDIN"))
+ .await;
+ assert!(matches!(result, Ok(None)), "{result:?}");
+ }
+
+ /// Proxy cannot tell whether a table or column it has not loaded is
+ /// encrypted, so with encrypted columns configured, a `COPY` into one is
+ /// refused, whatever the mapping errors setting.
+ #[tokio::test]
+ async fn copy_into_a_table_or_column_proxy_cannot_resolve_is_refused() {
+ for (sql, message) in [
+ ("COPY accounts FROM STDIN", unresolvable("accounts")),
+ (
+ "COPY archive.encrypted FROM STDIN",
+ unresolvable("archive.encrypted"),
+ ),
+ (
+ "CREATE TABLE IF NOT EXISTS late (secret text); COPY late FROM STDIN",
+ unresolvable("late"),
+ ),
+ (
+ "COPY notes (missing) FROM STDIN",
+ MappingError::CopyIntoUnknownColumn {
+ table: "notes".to_owned(),
+ column: "missing".to_owned(),
+ }
+ .to_string(),
+ ),
+ // The parser reads `CREATE TABLE t (LIKE u)` as a column named
+ // `LIKE`, so Proxy does not know the columns such a table has.
+ (
+ "CREATE TABLE copied (LIKE encrypted); COPY copied FROM STDIN",
+ unresolvable("copied"),
+ ),
+ ] {
+ let (mut frontend, _) = encrypted_client();
+ let forwarded = query(&mut frontend, sql).await;
+ assert!(is_refusal(&forwarded, &message), "{sql}: {forwarded}");
+ }
+ }
+
+ #[tokio::test]
+ async fn copy_is_forwarded_without_encrypted_columns() {
+ for sql in ["COPY accounts FROM STDIN", "COPY encrypted FROM STDIN"] {
+ let (mut frontend, _) = client(TandemConfig::for_testing(), false);
+ assert_eq!(query(&mut frontend, sql).await, sql, "{sql}");
+ }
+ }
+
+ /// A statement after a `COPY ... FROM STDIN` is a statement, not copy data,
+ /// so it is checked and encrypted like any other.
+ #[tokio::test]
+ async fn statements_after_copy_from_stdin_are_checked() {
+ let (mut frontend, _) = encrypted_client();
+ let forwarded = query(
+ &mut frontend,
+ "COPY notes FROM STDIN; INSERT INTO accounts (secret) VALUES ('plaintext')",
+ )
+ .await;
+ assert!(
+ is_refusal(&forwarded, &unresolvable("accounts")) && !forwarded.contains("plaintext"),
+ "{forwarded}"
+ );
+ }
+
+ /// Proxy writes out every statement again when it rewrites a Query, and a
+ /// `COPY` written out again is not the one the client sent, so a `COPY`
+ /// must be sent in a Query of its own.
+ #[tokio::test]
+ async fn copy_in_a_query_proxy_maps_is_refused() {
+ for sql in [
+ "COPY notes FROM STDIN; INSERT INTO encrypted (secret) VALUES ('plaintext')",
+ "INSERT INTO encrypted (secret) VALUES ('plaintext'); COPY notes TO STDOUT",
+ ] {
+ let (mut frontend, counter) = encrypted_client();
+ let forwarded = query(&mut frontend, sql).await;
+ assert!(
+ is_refusal(
+ &forwarded,
+ &MappingError::CopyWithEncryptedStatement.to_string()
+ ) && !forwarded.contains("plaintext"),
+ "{sql}: {forwarded}"
+ );
+ assert_eq!(counter.encrypted.load(Ordering::SeqCst), 0, "{sql}");
+ }
+ }
+
+ /// DDL that may change which columns are encrypted leaves the schema a
+ /// later `COPY` in the same Query is checked against unknown.
+ #[tokio::test]
+ async fn copy_after_ddl_that_may_change_encryption_is_refused() {
+ let (mut frontend, _) = encrypted_client();
+ let forwarded = query(
+ &mut frontend,
+ "CREATE TABLE IF NOT EXISTS encrypted (id int); COPY notes FROM STDIN",
+ )
+ .await;
+ assert!(
+ is_refusal(
+ &forwarded,
+ &MappingError::DependentStatementAfterDdl.to_string()
+ ),
+ "{forwarded}"
+ );
+ }
}
diff --git a/packages/cipherstash-proxy/src/postgresql/parser.rs b/packages/cipherstash-proxy/src/postgresql/parser.rs
index d2bf28e0..dc2ffa2b 100644
--- a/packages/cipherstash-proxy/src/postgresql/parser.rs
+++ b/packages/cipherstash-proxy/src/postgresql/parser.rs
@@ -3,7 +3,7 @@ use crate::prometheus::STATEMENTS_TOTAL;
use metrics::counter;
use sqltk::parser::ast;
use sqltk::parser::dialect::PostgreSqlDialect;
-use sqltk::parser::parser::Parser;
+use sqltk::parser::parser::{Parser, ParserError};
use sqltk::parser::tokenizer::{Span, Token, TokenWithSpan, Tokenizer, Whitespace};
const DIALECT: PostgreSqlDialect = PostgreSqlDialect {};
@@ -18,9 +18,17 @@ pub struct SqlParser;
impl SqlParser {
/// Parse a SQL statement string into an SqlParser AST
pub fn parse_statement(statement: &str) -> Result {
- let statement = Parser::new(&DIALECT)
- .with_tokens_with_locations(reset_tokens(statement)?)
- .parse_statement()?;
+ let tokens = reset_tokens(statement)?;
+ let statement = if contains_copy_from_stdin(&tokens) {
+ parse_copy_from_stdin_statements(tokens)?
+ .into_iter()
+ .next()
+ .ok_or_else(|| ParserError::ParserError("Expected: a statement".to_owned()))?
+ } else {
+ Parser::new(&DIALECT)
+ .with_tokens_with_locations(tokens)
+ .parse_statement()?
+ };
counter!(STATEMENTS_TOTAL).increment(1);
@@ -62,9 +70,14 @@ impl SqlParser {
/// Parse a SQL String potentially containing multiple statements into parsed SqlParser AST
pub fn parse_statements(statement: &str) -> Result, Error> {
- let statement = Parser::new(&DIALECT)
- .with_tokens_with_locations(reset_tokens(statement)?)
- .parse_statements()?;
+ let tokens = reset_tokens(statement)?;
+ let statement = if contains_copy_from_stdin(&tokens) {
+ parse_copy_from_stdin_statements(tokens)?
+ } else {
+ Parser::new(&DIALECT)
+ .with_tokens_with_locations(tokens)
+ .parse_statements()?
+ };
counter!(STATEMENTS_TOTAL).increment(statement.len() as u64);
@@ -80,7 +93,7 @@ impl SqlParser {
fn reset_tokens(sql: &str) -> Result, Error> {
let tokens = Tokenizer::new(&DIALECT, sql)
.tokenize_with_location()
- .map_err(|error| sqltk::parser::parser::ParserError::TokenizerError(error.to_string()))?;
+ .map_err(|error| ParserError::TokenizerError(error.to_string()))?;
// Built only once a RESET is found; until then the tokens are unchanged.
let mut rewritten: Option> = None;
let mut offset = 0;
@@ -93,7 +106,7 @@ fn reset_tokens(sql: &str) -> Result, Error> {
let body = &statement[..statement.len() - usize::from(has_delimiter)];
match leading_command(body) {
LeadingCommand::SetAll => {
- return Err(sqltk::parser::parser::ParserError::ParserError(
+ return Err(ParserError::ParserError(
"SET ALL is not valid PostgreSQL; use RESET ALL".to_owned(),
)
.into());
@@ -163,6 +176,65 @@ fn reset_tokens(sql: &str) -> Result, Error> {
Ok(rewritten.unwrap_or(tokens))
}
+/// The SQL parser expects the copy data of a `COPY ... FROM STDIN` to follow it
+/// in the SQL, as in a `psql` script: it refuses the statement without a `;`
+/// after it, and reads the rest of the SQL after one as copy data. A client
+/// sends the copy data in CopyData messages, so a statement after the `COPY` in
+/// a Query is a statement, which Proxy must check.
+fn contains_copy_from_stdin(tokens: &[TokenWithSpan]) -> bool {
+ tokens
+ .split(|token| token.token == Token::SemiColon)
+ .any(is_copy_from_stdin)
+}
+
+/// Whether `statement` is a `COPY` with `FROM` followed by `STDIN` anywhere in
+/// it. That also matches a query that reads a table named `stdin`, as in
+/// `COPY (SELECT * FROM stdin) TO STDOUT`, which only makes Proxy parse the
+/// Query one statement at a time, and so fails closed.
+fn is_copy_from_stdin(statement: &[TokenWithSpan]) -> bool {
+ let mut significant = statement
+ .iter()
+ .map(|token| &token.token)
+ .filter(|token| !matches!(token, Token::Whitespace(_)));
+ if !significant
+ .next()
+ .is_some_and(|token| is_keyword(token, "COPY"))
+ {
+ return false;
+ }
+ let mut after_from = false;
+ for token in significant {
+ if after_from && is_keyword(token, "STDIN") {
+ return true;
+ }
+ after_from = is_keyword(token, "FROM");
+ }
+ false
+}
+
+/// Parses each statement on its own, ending each `COPY ... FROM STDIN` with the
+/// `;` the SQL parser expects before its copy data, which is empty. A `;`
+/// splits statements wherever it is outside a literal, quoted identifier or
+/// comment, which is right for every statement but one with a body of several
+/// statements, such as a `BEGIN ATOMIC` function, which then fails to parse.
+fn parse_copy_from_stdin_statements(
+ tokens: Vec,
+) -> Result, ParserError> {
+ let mut statements = vec![];
+ for statement in tokens.split(|token| token.token == Token::SemiColon) {
+ let mut statement = statement.to_vec();
+ if is_copy_from_stdin(&statement) {
+ statement.push(TokenWithSpan::wrap(Token::SemiColon));
+ }
+ statements.extend(
+ Parser::new(&DIALECT)
+ .with_tokens_with_locations(statement)
+ .parse_statements()?,
+ );
+ }
+ Ok(statements)
+}
+
enum LeadingCommand {
/// `RESET name` or `RESET name.name`, by index of the RESET keyword and of
/// the last target token within the statement.
@@ -360,4 +432,53 @@ mod tests {
.map(|statement| SqlParser::statement_sql(&statement));
assert_ne!(rendered.as_deref(), Some("RESET ALL"));
}
+
+ /// A client sends the copy data of a `COPY ... FROM STDIN` in CopyData
+ /// messages, so the statements after it in the SQL are statements.
+ #[test]
+ fn statements_after_copy_from_stdin_are_statements() {
+ let sql = |statements: Vec| {
+ statements
+ .iter()
+ .map(|statement| match statement {
+ ast::Statement::Copy { values, .. } => {
+ assert!(values.is_empty());
+ "COPY".to_owned()
+ }
+ statement => SqlParser::statement_sql(statement),
+ })
+ .collect::>()
+ };
+
+ assert_eq!(
+ sql(SqlParser::parse_statements("COPY users (email) FROM STDIN").unwrap()),
+ vec!["COPY"]
+ );
+ assert_eq!(
+ sql(
+ SqlParser::parse_statements("COPY first FROM STDIN; COPY second FROM STDIN")
+ .unwrap()
+ ),
+ vec!["COPY", "COPY"]
+ );
+ assert_eq!(
+ sql(SqlParser::parse_statements(
+ "copy users from stdin with (format csv);\n INSERT INTO t VALUES ('a;b'); "
+ )
+ .unwrap()),
+ vec!["COPY", "INSERT INTO t VALUES ('a;b')"]
+ );
+ assert_eq!(
+ sql(vec![SqlParser::parse_statement(
+ "/* bulk */ COPY users FROM STDIN CSV"
+ )
+ .unwrap()]),
+ vec!["COPY"]
+ );
+ assert_eq!(
+ sql(SqlParser::parse_statements("COPY users FROM STDIN; RESET ALL").unwrap()),
+ vec!["COPY", "RESET ALL"]
+ );
+ assert!(SqlParser::parse_statements("COPY users FROM STDIN; SELECT 1 2").is_err());
+ }
}
diff --git a/packages/cipherstash-proxy/src/proxy/schema/middleware.rs b/packages/cipherstash-proxy/src/proxy/schema/middleware.rs
index 758f2623..3ff448f1 100644
--- a/packages/cipherstash-proxy/src/proxy/schema/middleware.rs
+++ b/packages/cipherstash-proxy/src/proxy/schema/middleware.rs
@@ -542,7 +542,7 @@ impl SchemaMiddleware {
let mut encryption_changing_ddl_seen = false;
for statement in statements {
- if encryption_changing_ddl_seen && eql_mapper::requires_type_check(statement) {
+ if encryption_changing_ddl_seen && eql_mapper::requires_schema(statement) {
return true;
}
encryption_changing_ddl_seen |= self.ddl_may_change_encryption(statement);
diff --git a/packages/eql-mapper/src/eql_mapper.rs b/packages/eql-mapper/src/eql_mapper.rs
index b3e2dddd..2221f1ec 100644
--- a/packages/eql-mapper/src/eql_mapper.rs
+++ b/packages/eql-mapper/src/eql_mapper.rs
@@ -95,6 +95,32 @@ pub fn writes_data(statement: &Statement) -> bool {
statement.accept(&mut FindWrite).is_break()
}
+/// Returns the table and column list a `COPY ... FROM` (a client stream, a file
+/// or a program) fills, or `None` for any other statement. An empty column list
+/// fills every column of the table.
+///
+/// A `COPY` is not type checked, so it is checked against the schema on its own.
+pub fn copies_into_table(statement: &Statement) -> Option<(&ast::ObjectName, &[ast::Ident])> {
+ match statement {
+ Statement::Copy {
+ source:
+ ast::CopySource::Table {
+ table_name,
+ columns,
+ },
+ to: false,
+ ..
+ } => Some((table_name, columns)),
+ _ => None,
+ }
+}
+
+/// Returns whether checking the [`Statement`] needs the schema: it is type
+/// checked, or it is a `COPY ... FROM` into a table.
+pub fn requires_schema(statement: &Statement) -> bool {
+ requires_type_check(statement) || copies_into_table(statement).is_some()
+}
+
/// The error type returned by various functions in the `eql_mapper` crate.
#[derive(Debug, PartialEq, Eq, thiserror::Error)]
pub enum EqlMapperError {
diff --git a/tests/python/tests/test_copy_from_stdin.py b/tests/python/tests/test_copy_from_stdin.py
new file mode 100644
index 00000000..ae0483e7
--- /dev/null
+++ b/tests/python/tests/test_copy_from_stdin.py
@@ -0,0 +1,47 @@
+# psycopg sends a COPY with no parameters as a simple Query.
+# Proxy forwards a COPY into a table with no encrypted columns, and refuses one into an encrypted column, as it does not encrypt copy data.
+import os
+import psycopg
+import pytest
+import random
+
+conn_params = {
+ "user": os.environ.get("CS_DATABASE__USERNAME"),
+ "password": os.environ.get("CS_DATABASE__PASSWORD"),
+ "dbname": os.environ.get("CS_DATABASE__NAME"),
+ "host": os.environ.get("CS_DATABASE__HOST"),
+ "port": 6432,
+}
+
+connection_str = psycopg.conninfo.make_conninfo(**conn_params)
+
+
+def make_id():
+ return random.randrange(1, 1000000000)
+
+
+def test_copy_into_a_plaintext_table():
+ first, second = make_id(), make_id()
+ with psycopg.connect(connection_str, autocommit=True) as conn:
+ with conn.cursor() as cursor:
+ with cursor.copy("COPY plaintext (id, plaintext) FROM STDIN") as copy:
+ copy.write_row((first, "alice"))
+ copy.write_row((second, "bob"))
+
+ cursor.execute(
+ "SELECT plaintext FROM plaintext WHERE id IN (%s, %s) ORDER BY plaintext",
+ [first, second],
+ )
+ assert [row[0] for row in cursor.fetchall()] == ["alice", "bob"]
+
+
+def test_copy_into_an_encrypted_column_is_refused():
+ with psycopg.connect(connection_str, autocommit=True) as conn:
+ with conn.cursor() as cursor:
+ with pytest.raises(psycopg.Error, match=r"COPY into encrypted column 'encrypted\.encrypted_text' is not supported"):
+ with cursor.copy("COPY encrypted (id, encrypted_text) FROM STDIN") as copy:
+ copy.write_row((make_id(), "alice@example.com"))
+
+ # The connection is still usable
+ cursor.execute("SELECT 1")
+ assert cursor.fetchone() == (1,)