diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c7fa584e..b4c0556a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - **Connection settings in transactions**: `SET CIPHERSTASH.KEYSET_ID`, `KEYSET_NAME` and `UNSAFE_DISABLE_MAPPING` now follow PostgreSQL's rules for settings in transactions. A change, including a `RESET`, `SET ... TO DEFAULT` or `DISCARD ALL`, takes effect once PostgreSQL confirms it, is kept on `COMMIT`, and is undone on `ROLLBACK` or `ROLLBACK TO SAVEPOINT`, or when the transaction fails, including the implicit transaction of a simple query or extended-protocol batch. `SET LOCAL` now lasts only until the end of the transaction. Previously Proxy kept these settings outside PostgreSQL's transactions: a rolled-back reset left Proxy with no keyset, so later writes were encrypted under `default_keyset_id` instead of the selected keyset; a rolled-back `SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING = true` left mapping disabled; and `SET LOCAL` was treated as a plain `SET`. Setting names are now matched as PostgreSQL matches them, ignoring case and quoting, and `UNSAFE_DISABLE_MAPPING` accepts every PostgreSQL boolean spelling. A `SET CIPHERSTASH.*` that Proxy cannot honour, such as an unknown setting name or a value it cannot interpret, is now refused with an error instead of being ignored. A simple query that Proxy rewrites now forwards every statement in it; previously statements Proxy did not map, such as a `SET`, could be dropped from a query that also encrypted values. A prepared statement is now refused with SQLSTATE `26000` if mapping has been turned on or off since it was prepared or bound, or if the keyset has changed and it holds values encrypted under the previous keyset, for example after a rollback undid the change it was prepared under; prepare it again to run it under the new settings. A statement or message that Proxy refuses now fails the batch, and an open transaction, in PostgreSQL as well, as an error from PostgreSQL would. Previously Proxy answered a refused extended-protocol batch itself, so statements earlier in the batch could still commit and the client's transaction carried on as if nothing had failed. A statement pipelined after a query or batch that may change these settings now waits for PostgreSQL to answer it, so it is mapped under the settings PostgreSQL confirmed. Previously Proxy assumed the change succeeded, so a `SET` that failed, or a `SET LOCAL` that ended with its implicit transaction, still applied to the statements sent after it. +- **`COPY ... FROM STDIN`**: Proxy now forwards a `COPY ... FROM STDIN` into a table with no encrypted columns, with its copy data, 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 or without an encryption configuration. A `COPY ... FROM` into an encrypted column is now refused with an error that names the column, as Proxy does not encrypt `COPY` data; previously a `COPY` from a file or program was forwarded with its plaintext. When encrypted columns are configured, a `COPY ... FROM` into a table or column that Proxy has not loaded, or into a table name that more than one schema has, is refused, as Proxy cannot tell whether it is encrypted. Statements after a `COPY ... FROM STDIN;` in the same simple query are now checked and encrypted like any other; previously Proxy read them as copy data and forwarded them unchanged, so a plaintext `INSERT` into an encrypted column could reach PostgreSQL. A simple query that has a `COPY` and a statement Proxy encrypts is refused, and the `COPY` must be sent in a query of its own. A `COPY (INSERT ... RETURNING ...) TO STDOUT`, or a `COPY` of another query that writes, is refused when encrypted columns are configured; previously Proxy forwarded its values as plaintext. Proxy now parses every `COPY`, so it refuses as invalid SQL the few `COPY` options its SQL parser does not support, such as `WHERE`, `HEADER MATCH` and `ON_ERROR`. + ## [2.2.4] - 2026-06-18 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index d2e0c6840..f095eb2a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -895,6 +895,7 @@ dependencies = [ "cipherstash-config", "cipherstash-proxy", "fake 4.2.0", + "futures-util", "hex", "postgres-types", "rand 0.9.2", diff --git a/docs/errors.md b/docs/errors.md index 78a97d753..66c3fd005 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -11,6 +11,10 @@ - [Invalid SQL statement](#mapping-invalid-sql-statement) - [Unsupported parameter type](#mapping-unsupported-parameter-type) - [Statement could not be type checked](#mapping-statement-could-not-be-type-checked) + - [COPY into encrypted column](#mapping-copy-into-encrypted-column) + - [COPY into unknown table](#mapping-copy-into-unknown-table) + - [COPY of a query that writes](#mapping-copy-of-query-that-writes) + - [COPY with a statement that uses an encrypted column](#mapping-copy-with-encrypted-statement) - [Internal Error](#mapping-internal-error) - [Prepared settings changed](#mapping-prepared-settings-changed) @@ -245,6 +249,107 @@ If the error persists, please contact CipherStash [support](https://cipherstash. + + + +## 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 send plaintext to the encrypted column. + +### Error message + +``` +COPY into encrypted column 'encrypted_email' of table 'users' is not supported, as Proxy does not encrypt COPY data. +``` + +### Notes + +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 a query that reads is forwarded, and returns the encrypted columns as ciphertext, as Proxy does not decrypt `COPY` data. +Proxy checks a `COPY` with the SQL parser, so it refuses as invalid SQL the few `COPY` options the parser does not support, such as `WHERE`, `HEADER MATCH` and `ON_ERROR`, even with no encryption configured. + +### 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 table + +A `COPY ... FROM` statement would load data into a table or column that is not in the database schema CipherStash Proxy has loaded. +Proxy cannot tell whether the data would go to an encrypted column, so it refuses the statement. + +### Error message + +``` +COPY into table 'users' is not supported, as Proxy cannot tell which of its columns are encrypted. +``` + +### Notes + +Proxy loads the tables of every schema by name, without the schema. +A table name used in more than one schema is refused, as Proxy cannot tell which table is meant. +A table created on the same connection, earlier or in the same query, is known by the name it was created with, so `CREATE TABLE public.staging` is known to `COPY public.staging` but not to `COPY staging`. +A `COPY` is refused if either a table created with that name or the loaded table of that name has an encrypted column it would fill, as Proxy cannot tell whether PostgreSQL created the table, such as after `CREATE TABLE IF NOT EXISTS` or a rollback. +So a `COPY` into a temporary table named like an encrypted table is refused. +A `COPY` with no column list into a table whose columns Proxy does not know, such as one created with `CREATE TABLE ... AS`, is refused. +Any `COPY` into a table created with `CREATE TABLE ... (LIKE ...)` is refused, as Proxy does not know which columns it copies. +When no columns are encrypted, Proxy does not check the table of a `COPY`. + +### How to fix + +Check that the table and columns exist. +If the table was created on another connection, retry once Proxy has reloaded the database schema. +If the table name is used in more than one schema, rename one of the tables. + + + + + +## COPY of a query that writes + +A `COPY (...) TO` statement has a query that writes, such as `INSERT ... RETURNING`, `UPDATE ... RETURNING` or `DELETE ... RETURNING`, or a `WITH` query that does. +CipherStash Proxy does not encrypt the query of a `COPY`, so it refuses the statement rather than send its values to PostgreSQL as plaintext. + +### Error message + +``` +COPY of a query that writes is not supported, as Proxy does not encrypt the query of a COPY. +``` + +### Notes + +When no columns are encrypted, Proxy forwards the `COPY`. + +### How to fix + +Run the statement on its own, without `COPY`, so Proxy encrypts it. + + + + + +## COPY with a statement that uses an encrypted column + +A simple query contains a `COPY` together with a statement that CipherStash Proxy encrypts. +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. + + diff --git a/packages/cipherstash-proxy-integration/Cargo.toml b/packages/cipherstash-proxy-integration/Cargo.toml index 82fe7bf6f..c4e6c48ab 100644 --- a/packages/cipherstash-proxy-integration/Cargo.toml +++ b/packages/cipherstash-proxy-integration/Cargo.toml @@ -10,6 +10,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 000000000..c1f560896 --- /dev/null +++ b/packages/cipherstash-proxy-integration/src/copy_from_stdin.rs @@ -0,0 +1,66 @@ +//! `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; + + let err = client + .copy_in::<_, Bytes>("COPY encrypted (id, encrypted_text) FROM STDIN") + .await + .err() + .unwrap(); + assert!(err.to_string().contains( + "COPY into encrypted column 'encrypted_text' of table 'encrypted' is not supported" + )); + + // 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!(err.to_string().contains("COPY into encrypted column")); + + // 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 8050045d0..758ecd323 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/src/error.rs b/packages/cipherstash-proxy/src/error.rs index b141b7367..f9b6a1f58 100644 --- a/packages/cipherstash-proxy/src/error.rs +++ b/packages/cipherstash-proxy/src/error.rs @@ -98,6 +98,18 @@ pub enum MappingError { #[error("Statement could not be type checked: {}. For help visit {}#mapping-statement-could-not-be-type-checked", _0, ERROR_DOC_BASE_URL)] StatementCouldNotBeTypeChecked(String), + #[error("COPY into encrypted column '{column}' of table '{table}' 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 table '{table}' is not supported, as Proxy cannot tell which of its columns are encrypted. For help visit {}#mapping-copy-into-unknown-table", ERROR_DOC_BASE_URL)] + CopyIntoUnknownTable { table: String }, + + #[error("COPY of a query that writes is not supported, as Proxy does not encrypt the query of a COPY. For help visit {}#mapping-copy-of-query-that-writes", ERROR_DOC_BASE_URL)] + CopyOfQueryThatWrites, + + #[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 could not be transformed: {0}")] StatementCouldNotBeTransformed(String), diff --git a/packages/cipherstash-proxy/src/postgresql/frontend.rs b/packages/cipherstash-proxy/src/postgresql/frontend.rs index de08cbbbe..81dcdee3f 100644 --- a/packages/cipherstash-proxy/src/postgresql/frontend.rs +++ b/packages/cipherstash-proxy/src/postgresql/frontend.rs @@ -513,9 +513,12 @@ where let mut encrypted = false; let mut parse_duration_recorded = false; - for (statement, (mapping_disabled, keyset)) in - parsed_statements.iter().zip(statement_settings) + for (index, (statement, (mapping_disabled, keyset))) in parsed_statements + .iter() + .zip(&statement_settings) + .enumerate() { + let mapping_disabled = *mapping_disabled; // Every statement is forwarded, so PostgreSQL answers each settings command if mapping_disabled { warn!(msg = "Encrypted statement mapping is not enabled"); @@ -526,6 +529,7 @@ where } self.check_for_schema_change(statement); + self.check_copy(statement)?; if !eql_mapper::requires_type_check(statement) { counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); @@ -540,7 +544,18 @@ where return Err(err); } else { // The Query is forwarded unchanged, so the settings commands after - // this statement still run, in order + // this statement still run, in order. A COPY after it is still checked. + for (statement, (mapping_disabled, _)) in parsed_statements + .iter() + .zip(&statement_settings) + .skip(index + 1) + { + // PostgreSQL stops at the failed statement, so a `CREATE` after it + // is not run, and is not recorded + if !mapping_disabled { + self.check_copy(statement)?; + } + } return Ok(None); }; } @@ -607,6 +622,17 @@ where }; } + // Proxy writes out each statement again when it rewrites a Query, and a `COPY` is not + // tested in a rewritten Query, so it must be sent in a Query of its own. Refused before the + // portal is added, so nothing is left to describe the results of a later Query. + if encrypted + && parsed_statements + .iter() + .any(|statement| matches!(statement, ast::Statement::Copy { .. })) + { + return Err(MappingError::CopyWithEncryptedStatement.into()); + } + // Record parse/typecheck duration (if not already recorded before encryption) if !parse_duration_recorded { self.context @@ -904,6 +930,7 @@ where } self.check_for_schema_change(&statement); + self.check_copy(&statement)?; if !eql_mapper::requires_type_check(&statement) { counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); @@ -1014,6 +1041,106 @@ where } } + /// Refuse a `COPY` that would send plaintext to an encrypted column, as Proxy does not + /// encrypt copy data: a `COPY ... FROM` into an encrypted column, and a `COPY` of a query + /// that writes, as Proxy does not type check the query of a `COPY ... TO`. Without a column + /// list, a `COPY ... FROM` fills every column. + /// + /// A `COPY ... FROM` into a table or column Proxy does not know, or into a table name that + /// more than one schema has, is refused, as Proxy cannot tell whether it is encrypted. With no + /// encrypted columns configured, every `COPY` is forwarded. + fn check_copy(&self, statement: &ast::Statement) -> Result<(), Error> { + let ast::Statement::Copy { source, to, .. } = statement else { + return Ok(()); + }; + if self.context.is_passthrough() { + return Ok(()); + } + match source { + ast::CopySource::Query(query) if writes(query) => { + Err(MappingError::CopyOfQueryThatWrites.into()) + } + ast::CopySource::Table { + table_name, + columns, + } if !to => self.check_copy_into(table_name, columns), + _ => Ok(()), + } + } + + /// Refuse a `COPY ... FROM` into `columns` of `table_name` unless Proxy knows none of them is + /// encrypted. + /// + /// The schema Proxy loads has the tables of every schema by name, without the schema. A table + /// created or changed on this connection is known by the name it was given. Proxy records it + /// before PostgreSQL runs the statement, which may then not create it, as with + /// `CREATE TABLE IF NOT EXISTS`, or roll it back. So the `COPY` is refused if either the table + /// created with that name or the loaded table of that name has an encrypted column it fills. + fn check_copy_into( + &self, + table_name: &ast::ObjectName, + columns: &[ast::Ident], + ) -> Result<(), Error> { + let unknown = || MappingError::CopyIntoUnknownTable { + table: table_name.to_string(), + }; + let resolver = self.context.get_table_resolver(); + let name = ast::ObjectName(table_name.0.last().cloned().into_iter().collect()); + let tables: Vec<_> = [ + resolver.resolve_table(table_name), + resolver.resolve_loaded_table(&name), + ] + .into_iter() + .flatten() + .collect(); + + // 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. + let created_like = tables.iter().any(|table| { + table.columns.iter().any(|column| { + column.name.quote_style.is_none() && column.name.value.eq_ignore_ascii_case("like") + }) + }); + if created_like { + return Err(unknown().into()); + } + + let mut targets = vec![]; + if columns.is_empty() { + for table in &tables { + targets.extend(table.columns.iter().map(|column| (table, column.clone()))); + } + } else { + for name in columns { + let found: Vec<_> = tables + .iter() + .filter_map(|table| Some((table, table.get_column(name).ok()?))) + .collect(); + if found.is_empty() { + return Err(unknown().into()); + } + targets.extend(found); + } + } + // A table with no columns Proxy knows, such as one created with `AS` + if targets.is_empty() { + return Err(unknown().into()); + } + + match targets + .iter() + .find(|(_, column)| matches!(column.kind, eql_mapper::ColumnKind::Eql(_))) + { + Some((table, column)) => Err(MappingError::CopyIntoEncryptedColumn { + table: table.name.value.clone(), + column: column.name.value.clone(), + } + .into()), + None => Ok(()), + } + } + /// The settings command `statement` runs. /// /// Refuses a `SET CIPHERSTASH.*` Proxy cannot honour: an unknown setting or a value it cannot @@ -1347,6 +1474,27 @@ where } } +/// Whether `query` writes: it is, or has a `WITH` query that is, an `INSERT`, `UPDATE` or +/// `DELETE`. PostgreSQL allows a statement that writes only there. +fn writes(query: &ast::Query) -> bool { + fn body_writes(body: &ast::SetExpr) -> bool { + match body { + ast::SetExpr::Insert(_) | ast::SetExpr::Update(_) | ast::SetExpr::Delete(_) => true, + ast::SetExpr::Query(query) => writes(query), + ast::SetExpr::SetOperation { left, right, .. } => { + body_writes(left) || body_writes(right) + } + ast::SetExpr::Select(_) | ast::SetExpr::Values(_) | ast::SetExpr::Table(_) => false, + } + } + query + .with + .iter() + .flat_map(|with| &with.cte_tables) + .any(|cte| writes(&cte.query)) + || body_writes(&query.body) +} + /// `Sync`, `Query` and `FunctionCall` are each answered with exactly one `ReadyForQuery` fn produces_ready_for_query(code: Code) -> bool { matches!(code, Code::Sync | Code::Query | Code::Unknown('F')) diff --git a/packages/cipherstash-proxy/src/postgresql/parser.rs b/packages/cipherstash-proxy/src/postgresql/parser.rs index 998e9062c..b7d512372 100644 --- a/packages/cipherstash-proxy/src/postgresql/parser.rs +++ b/packages/cipherstash-proxy/src/postgresql/parser.rs @@ -5,7 +5,8 @@ use sqltk::parser::ast; use sqltk::parser::dialect::PostgreSqlDialect; use sqltk::parser::keywords::Keyword; use sqltk::parser::parser::Parser; -use sqltk::parser::tokenizer::{Token, Tokenizer}; +use sqltk::parser::parser::ParserError; +use sqltk::parser::tokenizer::{Token, TokenWithSpan, Tokenizer}; const DIALECT: PostgreSqlDialect = PostgreSqlDialect {}; @@ -25,9 +26,17 @@ pub enum ResetStatement { impl SqlParser { /// Parse a SQL statement string into an SqlParser AST pub fn parse_statement(statement: &str) -> Result { - let statement = Parser::new(&DIALECT) - .try_with_sql(statement)? - .parse_statement()?; + let tokens = tokenize(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); @@ -36,9 +45,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) - .try_with_sql(statement)? - .parse_statements()?; + let tokens = tokenize(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); @@ -98,6 +112,64 @@ impl SqlParser { } } +fn tokenize(sql: &str) -> Result, ParserError> { + Ok(Tokenizer::new(&DIALECT, sql).tokenize_with_location()?) +} + +/// 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 `STDIN`, and parses 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) +} + +fn is_copy_from_stdin(tokens: &[TokenWithSpan]) -> bool { + let mut words = tokens + .iter() + .filter(|token| !matches!(token.token, Token::Whitespace(_))) + .map(|token| match &token.token { + Token::Word(word) => word.keyword, + _ => Keyword::NoKeyword, + }); + if words.next() != Some(Keyword::COPY) { + return false; + } + let mut after_from = false; + for keyword in words { + if after_from && keyword == Keyword::STDIN { + return true; + } + after_from = keyword == Keyword::FROM; + } + false +} + +/// Parse 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 tokens in tokens.split(|token| token.token == Token::SemiColon) { + let mut tokens = tokens.to_vec(); + if is_copy_from_stdin(&tokens) { + tokens.push(TokenWithSpan::wrap(Token::SemiColon)); + } + statements.extend( + Parser::new(&DIALECT) + .with_tokens_with_locations(tokens) + .parse_statements()?, + ); + } + Ok(statements) +} + #[cfg(test)] mod tests { use super::{ResetStatement, SqlParser}; @@ -152,4 +224,49 @@ mod tests { fn the_sql_parser_rejects_reset() { assert!(SqlParser::parse_statements("RESET ALL").is_err()); } + + /// 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 { + sqltk::parser::ast::Statement::Copy { values, .. } => { + assert!(values.is_empty()); + "COPY".to_owned() + } + statement => statement.to_string(), + }) + .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!(SqlParser::parse_statements("COPY users FROM STDIN; SELECT 1 2").is_err()); + } } diff --git a/packages/cipherstash-proxy/src/postgresql/tests.rs b/packages/cipherstash-proxy/src/postgresql/tests.rs index 7e6f931b5..9c783104f 100644 --- a/packages/cipherstash-proxy/src/postgresql/tests.rs +++ b/packages/cipherstash-proxy/src/postgresql/tests.rs @@ -16,11 +16,13 @@ use tokio::{ sync::mpsc, }; -/// Records the keyset of each encryption and decryption, and refuses it, as it has no keys +/// Records the keyset of each encryption and decryption, and refuses it, as it has no keys. +/// With `encrypts`, an encryption returns stub ciphertexts instead. #[derive(Clone, Default)] struct TestService { encrypted_with: Arc>>>, decrypted_with: Arc>>>, + encrypts: Arc, } #[async_trait::async_trait] @@ -29,12 +31,31 @@ impl EncryptionService for TestService { &self, keyset_id: Option, plaintexts: Vec>, - _columns: &[Option], + columns: &[Option], ) -> Result>, Error> { if plaintexts.iter().all(Option::is_none) { return Ok(vec![None; plaintexts.len()]); } self.encrypted_with.lock().unwrap().push(keyset_id); + if self.encrypts.load(std::sync::atomic::Ordering::Relaxed) { + // A ciphertext with no encrypted record, which is enough for Proxy to rewrite a + // statement + return Ok(plaintexts + .iter() + .zip(columns) + .map(|(plaintext, column)| { + plaintext.as_ref()?; + let identifier = &column.as_ref()?.identifier; + Some( + serde_json::from_value(serde_json::json!({ + "i": { "t": identifier.table, "c": identifier.column }, + "v": 2, + })) + .unwrap(), + ) + }) + .collect()); + } Err(crate::error::EncryptError::ColumnCouldNotBeEncrypted { table: "users".to_string(), column: "email".to_string(), @@ -75,6 +96,8 @@ struct Connection { encrypted_with: Arc>>>, /// The keyset of each decryption so far decrypted_with: Arc>>>, + /// Set to have encryptions return stub ciphertexts + encrypts: Arc, } impl Connection { @@ -94,10 +117,13 @@ impl Connection { "v": 1, "tables": { "users": { "email": {} } } })); + // Quoted column names, as Proxy loads them from the database let mut users = Table::new(Ident::new("users")); - users.add_column(Arc::new(eql_mapper::Column::native(Ident::new("id")))); + users.add_column(Arc::new(eql_mapper::Column::native(Ident::with_quote( + '"', "id", + )))); users.add_column(Arc::new(eql_mapper::Column::eql( - Ident::new("email"), + Ident::with_quote('"', "email"), EqlTraits::all(), ))); let mut schema = Schema::new("public"); @@ -113,6 +139,7 @@ impl Connection { let service = TestService::default(); let encrypted_with = service.encrypted_with.clone(); let decrypted_with = service.decrypted_with.clone(); + let encrypts = service.encrypts.clone(); let context = Context::new( 1, Arc::new(config), @@ -144,6 +171,7 @@ impl Connection { context, encrypted_with, decrypted_with, + encrypts, } } @@ -2061,3 +2089,481 @@ mod refusals { assert_eq!(connection.context.state_sizes().refusals, 0); } } + +/// A `COPY ... FROM STDIN` into a table with no encrypted columns is forwarded with its copy +/// data, and one into an encrypted column is refused, as Proxy does not encrypt copy data. +mod copy_from_stdin { + use super::*; + + fn copy_in_response() -> BytesMut { + // Text format, one text column + message(b'G', &[0, 0, 1, 0, 0]) + } + + fn copy_data(row: &str) -> BytesMut { + message(b'd', row.as_bytes()) + } + + fn copy_done() -> BytesMut { + message(b'c', b"") + } + + /// Check that Proxy refused what the client sent: it forwarded only a message PostgreSQL + /// rejects in its place, and none of the client's SQL. PostgreSQL then rejects it, and the + /// client gets Proxy's errors, which are returned. + async fn refused(connection: &mut Connection) -> Vec { + connection.errors_sent_to_client(); + let forwarded = connection.forwarded().await; + for (_, body) in &forwarded { + let body = String::from_utf8_lossy(body); + assert!(!body.contains("COPY") && !body.contains("alice"), "{body}"); + } + connection.server(rejection(&forwarded)).await; + connection.server(ready(b'I')).await; + + let mut errors = vec![]; + while let Ok(message) = connection.to_client.try_recv() { + if message.first() == Some(&b'E') { + let field = message[5..] + .split(|byte| *byte == 0) + .find(|field| field.first() == Some(&b'M')) + .unwrap(); + errors.push(String::from_utf8_lossy(&field[1..]).into_owned()); + } + } + errors + } + + /// The Query was forwarded unchanged + async fn forwarded_unchanged(connection: &mut Connection, sql: &str) { + assert_eq!( + connection.forwarded().await, + vec![(b'Q', format!("{sql}\0").into_bytes())] + ); + } + + #[tokio::test] + async fn a_simple_query_copy_into_a_plaintext_table_is_forwarded_with_its_data() { + let mut connection = Connection::started().await; + let sql = "COPY users (email) FROM STDIN"; + + connection.client(query(sql)).await; + connection.server(copy_in_response()).await; + connection.client(copy_data("alice@example.com\n")).await; + connection.client(copy_done()).await; + connection.server(command_complete("COPY 1")).await; + connection.server(ready(b'I')).await; + + assert_eq!( + connection.forwarded().await, + vec![ + (b'Q', format!("{sql}\0").into_bytes()), + (b'd', b"alice@example.com\n".to_vec()), + (b'c', vec![]), + ] + ); + assert_eq!(connection.errors_sent_to_client(), 0); + } + + #[tokio::test] + async fn a_simple_query_copy_into_an_encrypted_column_is_refused() { + let mut connection = Connection::encrypted().await; + + connection + .client(query("COPY users (id, email) FROM STDIN")) + .await; + + let errors = refused(&mut connection).await; + assert_eq!(errors.len(), 1); + assert!(errors[0] + .starts_with("COPY into encrypted column 'email' of table 'users' is not supported")); + } + + /// The SQL parser would read the statements after a `COPY ... FROM STDIN` as its copy data + #[tokio::test] + async fn a_statement_after_a_copy_in_one_simple_query_is_not_forwarded_as_plaintext() { + let mut connection = Connection::encrypted().await; + + connection + .client(query( + "COPY users (id) FROM STDIN; INSERT INTO users (email) VALUES ('alice@example.com')", + )) + .await; + + // The test encryption service refuses to encrypt the INSERT + assert_eq!(refused(&mut connection).await.len(), 1); + assert_eq!(connection.encrypted_with().len(), 1); + } + + /// Proxy writes out each statement again when it rewrites a Query to encrypt it + #[tokio::test] + async fn a_copy_in_one_simple_query_with_a_statement_that_is_encrypted_is_refused() { + let mut connection = Connection::encrypted().await; + connection + .encrypts + .store(true, std::sync::atomic::Ordering::Relaxed); + + connection + .client(query( + "INSERT INTO users (email) VALUES ('alice@example.com'); COPY users (id) FROM STDIN", + )) + .await; + + let errors = refused(&mut connection).await; + assert_eq!(errors.len(), 1); + assert!(errors[0].starts_with("COPY must be sent in a query of its own")); + } + + /// A refused Query leaves nothing behind to describe the results of the next one + #[tokio::test] + async fn the_query_after_a_refused_copy_with_an_encrypted_statement_is_not_decrypted() { + let mut connection = Connection::encrypted().await; + connection + .encrypts + .store(true, std::sync::atomic::Ordering::Relaxed); + connection + .client(query( + "INSERT INTO users (email) VALUES ('alice@example.com') RETURNING email; COPY users (id) FROM STDIN", + )) + .await; + refused(&mut connection).await; + + connection.client(query("SELECT 'hello'")).await; + let mut body = vec![]; + body.extend_from_slice(&1i16.to_be_bytes()); + body.extend_from_slice(&5i32.to_be_bytes()); + body.extend_from_slice(b"hello"); + let row = message(b'D', &body); + connection.server(row.clone()).await; + connection.server(command_complete("SELECT 1")).await; + connection.server(ready(b'I')).await; + + assert_eq!(connection.decrypted_with(), vec![]); + assert_eq!(connection.to_client.try_recv().unwrap(), row); + } + + #[tokio::test] + async fn an_extended_copy_into_a_plaintext_table_is_forwarded_with_its_data() { + let mut connection = Connection::started().await; + let sql = "COPY users (email) FROM STDIN"; + + connection.client(parse("", sql)).await; + connection.client(bind("", "")).await; + connection.client(execute("")).await; + connection.client(sync()).await; + connection.server(parse_complete()).await; + connection.server(bind_complete()).await; + connection.server(copy_in_response()).await; + connection.client(copy_data("alice@example.com\n")).await; + connection.client(copy_done()).await; + // PostgreSQL ignores the Sync sent during the copy + connection.client(sync()).await; + connection.server(command_complete("COPY 1")).await; + connection.server(ready(b'I')).await; + + assert_eq!(connection.forwarded_codes().await, b"PBESdcS".to_vec()); + assert_eq!(connection.errors_sent_to_client(), 0); + + // The next batch is matched to its own ReadyForQuery + connection.simple("SELECT 1", "SELECT 1").await; + assert_eq!(connection.errors_sent_to_client(), 0); + } + + #[tokio::test] + async fn an_extended_copy_into_an_encrypted_column_is_refused() { + let mut connection = Connection::encrypted().await; + + connection.client(parse("", "COPY users FROM STDIN")).await; + connection.client(bind("", "")).await; + connection.client(execute("")).await; + connection.client(sync()).await; + + let errors = refused(&mut connection).await; + assert_eq!(errors.len(), 1); + assert!(errors[0].starts_with("COPY into encrypted column 'email'")); + } + + /// PostgreSQL reads a file or program into the table as it reads copy data + #[tokio::test] + async fn a_copy_from_a_file_into_an_encrypted_column_is_refused() { + let mut connection = Connection::encrypted().await; + + connection + .client(query( + "COPY public.users FROM '/tmp/users.csv' WITH (FORMAT csv)", + )) + .await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into encrypted column 'email'")); + } + + #[tokio::test] + async fn a_copy_into_only_plaintext_columns_of_an_encrypted_table_is_forwarded() { + let mut connection = Connection::encrypted().await; + let sql = "COPY users (id) FROM STDIN"; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + + /// Proxy cannot tell whether a table it does not know has encrypted columns + #[tokio::test] + async fn a_copy_into_an_unknown_table_is_refused() { + let mut connection = Connection::encrypted().await; + + connection.client(query("COPY unknown FROM STDIN")).await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into table 'unknown' is not supported")); + } + + #[tokio::test] + async fn a_copy_into_a_table_created_in_the_same_query_is_forwarded() { + let mut connection = Connection::encrypted().await; + let sql = "CREATE TABLE staging (id int); COPY staging FROM STDIN"; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + + /// `CREATE TABLE ... AS` creates columns Proxy does not see + #[tokio::test] + async fn a_copy_with_no_column_list_into_a_table_with_no_known_columns_is_refused() { + let mut connection = Connection::encrypted().await; + + connection + .client(query( + "CREATE TABLE staging AS SELECT * FROM users; COPY staging FROM STDIN", + )) + .await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into table 'staging' is not supported")); + } + + /// `CREATE TABLE ... (LIKE ...)` copies columns Proxy does not see. The parser reads the + /// `LIKE` clause as a column named `LIKE`. + #[tokio::test] + async fn a_copy_into_a_table_created_like_another_is_refused() { + for sql in [ + "CREATE TABLE staging (LIKE users); COPY staging FROM STDIN", + "CREATE TABLE staging (like users, note text); COPY staging FROM STDIN", + "CREATE TABLE staging (LIKE users, note text); COPY staging (note) FROM STDIN", + ] { + let mut connection = Connection::encrypted().await; + + connection.client(query(sql)).await; + + let errors = refused(&mut connection).await; + assert!( + errors[0].starts_with("COPY into table 'staging' is not supported"), + "{sql}" + ); + } + } + + /// PostgreSQL finds a table named without a schema in the temporary schema first, but a + /// schema-qualified name is the table in that schema + #[tokio::test] + async fn a_schema_qualified_copy_is_not_hidden_by_a_table_of_the_same_name() { + let mut connection = Connection::encrypted().await; + + connection + .client(query( + "CREATE TEMP TABLE users (id int, email text); COPY public.users FROM STDIN", + )) + .await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into encrypted column 'email' of table 'users'")); + } + + #[tokio::test] + async fn a_schema_qualified_copy_into_a_table_created_with_that_name_is_forwarded() { + let mut connection = Connection::encrypted().await; + let sql = "CREATE TABLE public.staging (id int); COPY public.staging FROM STDIN"; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + + /// The Query is forwarded unchanged after a statement that fails type checking + #[tokio::test] + async fn a_copy_after_a_statement_that_fails_type_checking_is_still_checked() { + let mut connection = Connection::encrypted().await; + + connection + .client(query("SELECT * FROM unknown; COPY users FROM STDIN")) + .await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into encrypted column 'email'")); + } + + /// The query of a `COPY ... TO` is not type checked, so one that writes would send its + /// values to PostgreSQL as plaintext + #[tokio::test] + async fn a_copy_of_a_query_that_writes_is_refused() { + for sql in [ + "COPY (INSERT INTO users (email) VALUES ('alice@example.com') RETURNING id) TO STDOUT", + "COPY (UPDATE users SET email = 'alice@example.com' RETURNING id) TO STDOUT", + "COPY (DELETE FROM users WHERE email = 'alice@example.com' RETURNING id) TO STDOUT", + "COPY (WITH i AS (INSERT INTO users (email) VALUES ('alice@example.com') RETURNING id) SELECT id FROM i) TO STDOUT", + "COPY ((WITH i AS (INSERT INTO users (email) VALUES ('alice@example.com') RETURNING id) SELECT id FROM i)) TO STDOUT", + ] { + let mut connection = Connection::encrypted().await; + + connection.client(query(sql)).await; + + let errors = refused(&mut connection).await; + assert!( + errors[0].starts_with("COPY of a query that writes is not supported"), + "{sql}: {errors:?}" + ); + } + } + + /// Proxy records a table created on this connection before PostgreSQL runs the `CREATE`, which + /// may then not create it + #[tokio::test] + async fn a_copy_into_an_encrypted_table_is_not_hidden_by_a_table_created_on_the_connection() { + let mut connection = Connection::encrypted().await; + connection + .client(query( + "CREATE TABLE IF NOT EXISTS users (id int, email text); COPY users FROM STDIN", + )) + .await; + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into encrypted column 'email'")); + + let mut connection = Connection::encrypted().await; + connection + .client(query( + "CREATE TEMP TABLE users (id int, email text); COPY unknown FROM STDIN", + )) + .await; + refused(&mut connection).await; + connection.client(query("COPY users FROM STDIN")).await; + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into encrypted column 'email'")); + } + + /// PostgreSQL stops at the statement that fails, so it does not run the `CREATE` after it + #[tokio::test] + async fn a_table_created_after_a_statement_that_fails_type_checking_is_not_known() { + let mut connection = Connection::encrypted().await; + + connection + .client(query( + "SELECT * FROM unknown; CREATE TABLE staging (id int); COPY staging FROM STDIN", + )) + .await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into table 'staging' is not supported")); + } + + /// The loaded schema has tables by name, without their schema + #[tokio::test] + async fn a_copy_into_a_table_name_two_schemas_have_is_refused() { + let mut config = TandemConfig::for_testing(); + config.encrypt.default_keyset_id = None; + let encrypt_config = EncryptConfig::from_json(serde_json::json!({ + "v": 1, + "tables": { "users": { "email": {} } } + })); + let mut plaintext = Table::new(Ident::new("users")); + plaintext.add_column(Arc::new(eql_mapper::Column::native(Ident::with_quote( + '"', "email", + )))); + let mut encrypted = Table::new(Ident::new("users")); + encrypted.add_column(Arc::new(eql_mapper::Column::eql( + Ident::with_quote('"', "email"), + EqlTraits::all(), + ))); + let mut schema = Schema::new("public"); + schema.add_table(plaintext); + schema.add_table(encrypted); + let mut connection = Connection::with_schema(config, encrypt_config, schema); + connection.server(ready(b'I')).await; + + connection.client(query("COPY users FROM STDIN")).await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into table 'users' is not supported")); + } + + #[tokio::test] + async fn a_copy_into_an_unknown_column_of_a_known_table_is_refused() { + let mut connection = Connection::encrypted().await; + + connection + .client(query("COPY users (id, nickname) FROM STDIN")) + .await; + + let errors = refused(&mut connection).await; + assert!(errors[0].starts_with("COPY into table 'users' is not supported")); + } + + /// Only an unquoted `LIKE` is the clause + #[tokio::test] + async fn a_copy_into_a_table_with_a_quoted_like_column_is_forwarded() { + let mut connection = Connection::encrypted().await; + let sql = r#"CREATE TABLE staging ("like" text); COPY staging FROM STDIN"#; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + + #[tokio::test] + async fn a_copy_into_an_encrypted_column_is_forwarded_with_mapping_disabled() { + for sql in [ + "SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING = true; COPY users FROM STDIN", + "SELECT * FROM unknown; SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING = true; COPY users FROM STDIN", + ] { + let mut connection = Connection::encrypted().await; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + } + + /// Proxy returns the ciphertext, as it does not decrypt copy data + #[tokio::test] + async fn a_copy_of_an_encrypted_table_to_stdout_is_forwarded() { + let mut connection = Connection::encrypted().await; + let sql = "COPY users TO STDOUT"; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + + /// Proxy returns the ciphertext, as it does not decrypt copy data + #[tokio::test] + async fn a_copy_to_stdout_of_a_query_that_only_reads_is_forwarded() { + let mut connection = Connection::encrypted().await; + let sql = "COPY (SELECT id, email FROM users) TO STDOUT"; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } + + #[tokio::test] + async fn a_copy_of_a_query_that_writes_is_forwarded_with_no_encrypted_columns() { + let mut connection = Connection::started().await; + let sql = + "COPY (INSERT INTO users (email) VALUES ('alice@example.com') RETURNING id) TO STDOUT"; + + connection.client(query(sql)).await; + + forwarded_unchanged(&mut connection, sql).await; + } +} diff --git a/packages/eql-mapper/src/model/schema_delta.rs b/packages/eql-mapper/src/model/schema_delta.rs index 24f7d1903..a9e94fcd6 100644 --- a/packages/eql-mapper/src/model/schema_delta.rs +++ b/packages/eql-mapper/src/model/schema_delta.rs @@ -67,6 +67,13 @@ impl SchemaWithEdits { } } + pub(crate) fn resolve_loaded_table( + &self, + name: &ObjectName, + ) -> Result, SchemaError> { + self.schema.resolve_table(name) + } + pub(crate) fn resolve_table_columns( &self, table_name: &ObjectName, diff --git a/packages/eql-mapper/src/model/table_resolver.rs b/packages/eql-mapper/src/model/table_resolver.rs index e461ec488..0e853b0ce 100644 --- a/packages/eql-mapper/src/model/table_resolver.rs +++ b/packages/eql-mapper/src/model/table_resolver.rs @@ -44,6 +44,16 @@ impl TableResolver { } } + /// Resolves a table in the loaded schema, ignoring any edits made since it was loaded. + pub fn resolve_loaded_table(&self, name: &ObjectName) -> Result, SchemaError> { + match self { + TableResolver::ViaSchema(schema) => schema.resolve_table(name), + TableResolver::ViaSchemaWithEdits(schema_with_edits) => { + schema_with_edits.read().unwrap().resolve_loaded_table(name) + } + } + } + pub fn resolve_table_columns( &self, table_name: &ObjectName, 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 000000000..263d848ac --- /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_text' of table 'encrypted' 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,)