diff --git a/CHANGELOG.md b/CHANGELOG.md index 45a32725..65a4e470 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Fixed +- **Schema-qualified table names**: Proxy now resolves a schema-qualified name such as `public.users` to the table in that schema, so statements that use one are encrypted instead of failing type checking. This includes a table whose name is also used in another schema on the search path. Proxy keys encryption configuration by the unqualified table name, so when same-named tables in different schemas declare a column with different EQL domains, statements on that column are refused rather than encrypted with the other table's configuration. Same-named tables that declare the same domain share one configuration and key descriptor, so their ciphertexts and search terms are interchangeable. + - **Prepared statement recovery and settings isolation**: failed or discarded protocol messages no longer replace encryption metadata for statements PostgreSQL retained. Pipelined close-and-reprepare recovers immediately, and refused operations use names that remain unpredictable within PostgreSQL's lookup limit. Unsupported keyset values fail explicitly, and stale prepared statements signal driver cache invalidation. - **Ordinary transaction and COPY traffic**: transaction and reset batches that do not change effective CipherStash settings remain usable, and COPY input can progress while transaction outcomes are pending. Settings tracking avoids unnecessary transaction barriers and repeated SQL tree copies. @@ -69,6 +71,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - **Bind to a statement Proxy did not see prepared**: Proxy now refuses a Bind that names a prepared statement it has no Parse for on the connection, with PostgreSQL's own `26000` error, and the connection stays open. Proxy keeps encryption details only for statements that touch encrypted columns, so before this change it could not tell an unknown statement from a plaintext one and sent the bound values as they were. This includes statements created with SQL `PREPARE` and then bound through the extended protocol; prepare them with Parse instead. The check does not apply while mapping is disabled. See [Unknown prepared statement](docs/errors.md#mapping-unknown-prepared-statement). +- **Writes to a table Proxy cannot resolve**: when encrypted columns are configured, Proxy now refuses an `INSERT`, `UPDATE`, `DELETE` or `MERGE` that refers to a table missing from the database schema it has loaded, whether or not `CS_DEVELOPMENT__ENABLE_MAPPING_ERRORS` is set. Previously such a statement failed type checking and was forwarded as written, so its values could reach PostgreSQL in plaintext. A simple query is also no longer forwarded as written when one statement fails type checking and another needs encrypting. See [Unresolvable table](docs/errors.md#mapping-unresolvable-table). + + Compatibility note: while encrypted columns are configured, Proxy now refuses writes to tables in schemas outside the search path it loads, to views and foreign tables (Proxy loads only base tables), and writes that use a bare table name found in more than one schema on that search path. Before upgrading, check that applications do not write to such tables through Proxy, or qualify the table name with its schema. + - **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/docs/errors.md b/docs/errors.md index 02a91c87..b7c6023f 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -17,6 +17,7 @@ - [Prepared settings changed](#mapping-prepared-settings-changed) - [Unmodelled DDL](#mapping-unmodelled-ddl) - [Unmappable encrypted column](#mapping-unmappable-encrypted-column) + - [Unresolvable table](#mapping-unresolvable-table) - [Unknown prepared statement](#mapping-unknown-prepared-statement) - [Internal Error](#mapping-internal-error) @@ -404,6 +405,45 @@ If the column is no longer used, dropping it also resolves the error. + + + +## Unresolvable table + +A statement that writes data refers to a table that is not in the database schema CipherStash Proxy has loaded. +Proxy cannot tell whether the statement writes to an encrypted column, so it refuses the statement rather than forward its values unencrypted. + +### Error message + +``` +Statement refers to table 'users', which is not in the database schema Proxy has loaded, so Proxy cannot tell whether it writes to an encrypted column. +``` + +### Notes + +The refusal applies to `INSERT`, `UPDATE`, `DELETE` and `MERGE`, including one inside `PREPARE`, `EXPLAIN` or a `WITH` clause, when any column is encrypted. +It applies wherever the statement names the table, so `INSERT INTO users ... SELECT ... FROM other` is refused when `other` cannot be resolved. +Read-only statements are not refused. +When no columns are encrypted, Proxy forwards these statements. + +Unlike a type-check failure, this refusal is not subject to the `CS_DEVELOPMENT__ENABLE_MAPPING_ERRORS` passthrough. + +Proxy loads the tables in the schemas on its database connection's search path. +A schema-qualified name such as `public.users` resolves only to a table loaded from that schema. +A bare table name used in more than one schema on the search path cannot be resolved, because the client's search path can differ from Proxy's. +Encryption configuration is keyed by the unqualified table name, so when same-named tables in different schemas declare a column with different EQL domains, Proxy loads no configuration for that column and refuses statements on it. +When they declare the same domain, both tables share one configuration and one key descriptor, so they are not cryptographically separate: a ciphertext copied from one table to the other still decrypts, and equal values produce equal search terms in both. +A table in a schema that is not on the search path, or one created on another connection since Proxy last loaded the schema, cannot be resolved. + +### How to fix + +Check that the table exists and that its name is spelled as PostgreSQL resolves it. +If the table name is used in more than one schema, qualify it with its schema, as in `public.users`. +If the table is in a schema outside the search path, add that schema to the search path of the database user Proxy connects as, then restart Proxy or wait for it to reload the schema. +If the table was created on another connection, retry once Proxy has reloaded the database schema. + + + diff --git a/packages/cipherstash-proxy/CONTEXT.md b/packages/cipherstash-proxy/CONTEXT.md index ac7b1649..2f678bf6 100644 --- a/packages/cipherstash-proxy/CONTEXT.md +++ b/packages/cipherstash-proxy/CONTEXT.md @@ -54,9 +54,12 @@ _Avoid_: session (`start_session`, `SessionId` and the `..._statements_session_duration_seconds` metric all use this sense and are misnamed). **Passthrough**: -Traffic forwarded without encryption or rewriting. Three independent conditions produce +Traffic forwarded without encryption or rewriting. Four independent conditions produce it and they are not interchangeable: no encrypt config is loaded, mapping is disabled by -configuration, or `SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING` was issued on this connection. +configuration, `SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING` was issued on this connection, or +the **type-check fallback** forwarded a statement EQL Mapper could not type check while +mapping errors are off. The fallback never applies to a refusal, and never to a simple +Query with another statement that needs mapping. Name the condition rather than saying "passthrough" alone. ### Encryption diff --git a/packages/cipherstash-proxy/src/error.rs b/packages/cipherstash-proxy/src/error.rs index 83463623..532cbc5b 100644 --- a/packages/cipherstash-proxy/src/error.rs +++ b/packages/cipherstash-proxy/src/error.rs @@ -73,9 +73,11 @@ impl Error { self, // Forwarding a statement that references a legacy EQL v2 column // stores plaintext in a column its operator believes is encrypted - // (CIP-3688). No configuration may turn that back on. + // (CIP-3688). No configuration may turn that back on. Nor may a + // write to a table Proxy cannot resolve be forwarded. Error::Mapping( MappingError::UnmappableEncryptedColumn { .. } + | MappingError::UnresolvableTable { .. } | MappingError::DependentStatementAfterDdl | MappingError::DependentStatementAfterSettings | MappingError::PreparedSettingsChanged @@ -153,6 +155,9 @@ 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("Statement refers to table '{table}', which is not in the database schema Proxy has loaded, so Proxy cannot tell whether it writes to an encrypted column. For help visit {}#mapping-unresolvable-table", ERROR_DOC_BASE_URL)] + UnresolvableTable { table: String }, + #[error("Column '{table}.{column}' is declared as '{column_type}', a legacy EQL v2 type that this version of CipherStash Proxy cannot encrypt or decrypt. Statements referencing '{table}' are refused so that plaintext is never written to the column. Migrate '{table}.{column}' to an EQL v3 domain type. For help visit {}#mapping-unmappable-encrypted-column", ERROR_DOC_BASE_URL)] UnmappableEncryptedColumn { table: String, diff --git a/packages/cipherstash-proxy/src/postgresql/context/mod.rs b/packages/cipherstash-proxy/src/postgresql/context/mod.rs index a418ad83..c41a00e6 100644 --- a/packages/cipherstash-proxy/src/postgresql/context/mod.rs +++ b/packages/cipherstash-proxy/src/postgresql/context/mod.rs @@ -1732,6 +1732,30 @@ where Ok(()) } + /// Records the native tables a simple-query batch creates, so the batch's + /// later statements can write to them. + pub fn note_native_tables(&self, statements: &[sqltk::parser::ast::Statement]) { + self.schema_middleware.note_native_tables(statements); + } + + /// Whether every table `statement` writes to is a native table this + /// connection created that the schema may not model yet. + pub fn writes_only_native_tables(&self, statement: &sqltk::parser::ast::Statement) -> bool { + eql_mapper::write_targets(statement).is_some_and(|targets| { + !targets.is_empty() + && targets + .iter() + .all(|target| self.schema_middleware.is_native_table(target)) + }) + } + + /// Whether any column is encrypted. A column can be encrypted without + /// encrypt config, when same-named tables disagree on its domain. + pub fn has_encrypted_columns(&self) -> bool { + !self.schema_middleware.encrypt_config().is_empty() + || self.schema_middleware.schema_has_encrypted_columns() + } + pub fn is_passthrough(&self) -> bool { self.schema_middleware.encrypt_config().is_empty() || self.config.mapping_disabled() } diff --git a/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs b/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs index da2dcf7c..88933e44 100644 --- a/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs +++ b/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs @@ -369,6 +369,9 @@ impl Frontend { { return Err(MappingError::DependentStatementAfterDdl.into()); } + // So a write later in this batch to a native table it creates is not + // refused as a write to an unresolvable table. + self.context.note_native_tables(&parsed_statements); if parsed_statements .iter() .any(eql_mapper::requires_type_check) @@ -385,9 +388,10 @@ impl Frontend { let mut portal = Portal::passthrough(Some(session_id)); let mut encrypted = false; + let mut mapped = false; let mut parse_duration_recorded = false; - for statement in &parsed_statements { + for (index, statement) in parsed_statements.iter().enumerate() { if self.context.unsafe_disable_mapping()? { warn!(msg = "Encrypted statement mapping is not enabled"); counter!(STATEMENTS_PASSTHROUGH_MAPPING_DISABLED_TOTAL).increment(1); @@ -412,6 +416,13 @@ impl Frontend { // fallback altogether. if self.context.mapping_errors_enabled() || err.must_fail_closed() { return Err(err); + } + // The fallback forwards the whole Query as the client wrote + // it, so it must not carry another statement that needs + // mapping: that statement's values would reach the database + // unencrypted. + if mapped || self.any_requires_mapping(&parsed_statements[index + 1..])? { + return Err(err); } else { self.context.register_query( operation, @@ -430,6 +441,7 @@ impl Frontend { client_id = self.context.client_id, msg = "Encryptable Statement", ); + mapped = true; let mut transformed = false; if typed_statement.requires_transform() { @@ -1216,6 +1228,30 @@ impl Frontend { Ok(()) } + /// Whether any of `statements` touches an encrypted column. A statement + /// that must be refused returns its refusal; one that fails type checking + /// for another reason counts as not touching one. + fn any_requires_mapping(&self, statements: &[ast::Statement]) -> Result { + for statement in statements { + if !eql_mapper::requires_type_check(statement) { + continue; + } + match self.type_check(statement) { + Ok(typed_statement) => { + if self + .to_encryptable_statement(&typed_statement, vec![])? + .is_some() + { + return Ok(true); + } + } + Err(err) if err.must_fail_closed() => return Err(err), + Err(_) => {} + } + } + Ok(false) + } + fn type_check<'a>( &self, statement: &'a ast::Statement, @@ -1249,6 +1285,26 @@ impl Frontend { } .into()) } + // The type checker never saw the table's columns, so a write to it + // could put plaintext in an encrypted column. + Err(err) + if err.as_table_not_found().is_some() + && eql_mapper::writes_data(statement) + && self.context.has_encrypted_columns() + && !self.context.writes_only_native_tables(statement) => + { + let table = err.as_table_not_found().unwrap(); + warn!( + client_id = self.context.client_id, + msg = "Statement refused: it writes data and refers to a table that is not in the loaded database schema, so Proxy cannot tell whether it writes to an encrypted column.", + table, + ); + counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); + Err(MappingError::UnresolvableTable { + table: table.to_string(), + } + .into()) + } Err(EqlMapperError::InternalError(str)) => { warn!( client_id = self.context.client_id, @@ -3570,3 +3626,485 @@ mod settings_reset_tests { assert_eq!(frontend.context.keyset_identifier().unwrap(), None); } } + +/// A statement Proxy cannot type check is forwarded as written when +/// mapping errors are off. These pin when that fallback is refused instead. +#[cfg(test)] +mod unresolvable_table_tests { + use super::Frontend; + use crate::config::TandemConfig; + use crate::error::{Error, MappingError}; + use crate::postgresql::context::{Context, KeysetIdentifier}; + use crate::postgresql::test_operation_id as operation_id; + use crate::postgresql::Column; + use crate::proxy::{EncryptConfig, EncryptionService}; + use bytes::Bytes; + use cipherstash_client::encryption::Plaintext; + use cipherstash_client::eql::{EncryptedPayloadV3, EQL_SCHEMA_VERSION_V3}; + use cipherstash_client::zerokms::EncryptedRecord; + use pg_proto::{FrontendMessage, FrontendMiddlewareOutput, Parse}; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + + /// Encrypts every value as a stub ciphertext and counts the values. + #[derive(Clone, Default)] + struct Counter { + encrypted: Arc, + } + + #[async_trait::async_trait] + impl EncryptionService for Counter { + async fn encrypt( + &self, + _: Option, + plaintexts: Vec>, + _: &[Option], + ) -> Result>, Error> { + self.encrypted.fetch_add(plaintexts.len(), Ordering::SeqCst); + Ok(plaintexts + .iter() + .map(|_| Some(crate::EqlOutput::Store(ciphertext()))) + .collect()) + } + async fn decrypt( + &self, + _: Option, + _: Vec>, + ) -> Result>, Error> { + Ok(Vec::new()) + } + async fn decrypt_inbound_eql( + &self, + _: Option, + _: Vec>, + ) -> Result>, Error> { + Ok(Vec::new()) + } + } + + fn ciphertext() -> crate::EqlCiphertext { + crate::EqlCiphertext::Encrypted(EncryptedPayloadV3 { + version: EQL_SCHEMA_VERSION_V3, + identifier: crate::Identifier::new("encrypted", "secret"), + ciphertext: EncryptedRecord { + iv: Default::default(), + ciphertext: vec![1; 16], + tag: vec![2; 16], + descriptor: "encrypted/secret".into(), + keyset_id: Some(uuid::Uuid::nil()), + decryption_policy: None, + }, + hmac_256: None, + bloom_filter: None, + ore_block_u64_8_256: None, + ope_cllw: None, + }) + } + + /// A connection whose schema has `encrypted (id, secret)` with `secret` + /// encrypted, and an `encrypted` encrypt config only when `encrypted` is set. + fn client(config: TandemConfig, encrypted: bool) -> (Frontend, Counter) { + let mut schema = eql_mapper::Schema::new("public"); + let mut table = eql_mapper::Table::new(sqltk::parser::ast::Ident::new("encrypted")); + table.add_column(Arc::new(eql_mapper::Column::native( + sqltk::parser::ast::Ident::new("id"), + ))); + let secret = sqltk::parser::ast::Ident::new("secret"); + table.add_column(Arc::new(if encrypted { + eql_mapper::Column::eql( + secret, + eql_mapper::EqlTraits::none(), + eql_mapper::DomainIdentity::from_domain_name("eql_v3_text").unwrap(), + ) + } else { + eql_mapper::Column::native(secret) + })); + schema.add_table(table); + let mut encryption = EncryptConfig::new(); + if encrypted { + encryption.insert( + crate::Identifier::new("encrypted", "secret"), + cipherstash_client::schema::ColumnConfig::build("secret".to_owned()) + .casts_as(cipherstash_client::schema::ColumnType::Text), + ); + } + let counter = Counter::default(); + let (sender, _) = tokio::sync::mpsc::unbounded_channel(); + let context = Context::new( + 1, + Arc::new(config), + Arc::new(encryption), + Arc::new(schema), + Arc::new(rustls::RootCertStore::empty()), + counter.clone(), + sender, + ); + (Frontend::new(context), counter) + } + + fn encrypted_client() -> (Frontend, Counter) { + client(TandemConfig::for_testing(), true) + } + + /// Sends a simple Query and returns the SQL Proxy forwards for it. + async fn query(frontend: &mut Frontend, sql: &str) -> String { + let output = frontend + .intercept( + operation_id(), + FrontendMessage::Query(Bytes::copy_from_slice(sql.as_bytes())), + ) + .await + .unwrap(); + match output { + FrontendMiddlewareOutput::Forward(FrontendMessage::Query(query)) + | FrontendMiddlewareOutput::ForwardThenFlush(FrontendMessage::Query(query)) => { + String::from_utf8_lossy(&query).into_owned() + } + other => panic!("expected a forwarded query, got {other:?}"), + } + } + + fn unresolvable(table: &str) -> String { + MappingError::UnresolvableTable { + table: table.to_owned(), + } + .to_string() + } + + fn assert_refused(forwarded: &str, table: &str) { + // The message is quoted once in the DO block's body, then again as + // the DO block's literal. + assert!( + forwarded.contains("RAISE EXCEPTION") + && forwarded.contains(&unresolvable(table).replace('\'', "''''")), + "expected a refusal naming {table}, forwarded: {forwarded}" + ); + assert!( + !forwarded.contains("plaintext"), + "the plaintext reached PostgreSQL: {forwarded}" + ); + } + + const WRITES: [(&str, &str); 6] = [ + ( + "INSERT INTO accounts (secret) VALUES ('plaintext')", + "accounts", + ), + ( + "UPDATE accounts SET secret = 'plaintext' WHERE id = 1", + "accounts", + ), + ("DELETE FROM accounts WHERE secret = 'plaintext'", "accounts"), + ( + "INSERT INTO archive.encrypted (secret) VALUES ('plaintext')", + "archive.encrypted", + ), + ( + "INSERT INTO encrypted (id, secret) SELECT id, 'plaintext' FROM accounts", + "accounts", + ), + ( + "MERGE INTO accounts USING encrypted ON accounts.id = encrypted.id WHEN NOT MATCHED THEN INSERT (secret) VALUES ('plaintext')", + "accounts", + ), + ]; + + #[tokio::test] + async fn simple_write_on_an_unresolvable_table_is_refused() { + for (sql, table) in WRITES { + let (mut frontend, _) = encrypted_client(); + assert_refused(&query(&mut frontend, sql).await, table); + } + } + + #[tokio::test] + async fn extended_write_on_an_unresolvable_table_is_refused() { + for (sql, table) in WRITES { + let (mut frontend, _) = encrypted_client(); + let result = frontend + .parse_handler( + operation_id(), + Parse { + statement: Bytes::from_static(b"s"), + query: Bytes::copy_from_slice(sql.as_bytes()), + parameter_types: Vec::new(), + }, + ) + .await; + assert!( + matches!( + &result, + Err(Error::Mapping(MappingError::UnresolvableTable { table: refused })) + if refused == table + ), + "{sql}: {result:?}" + ); + } + } + + /// The schema middleware does not model a native temporary table, but + /// guarantees it holds no encrypted column and shadows no encrypted table, + /// so a write to one is not refused (the Go pgx integration test). + #[tokio::test] + async fn write_to_a_native_temporary_table_is_forwarded() { + let (mut frontend, _) = encrypted_client(); + let sql = "CREATE TEMPORARY TABLE t (name text); INSERT INTO t (name) VALUES ('Ada')"; + assert_eq!(query(&mut frontend, sql).await, sql); + + let (mut frontend, _) = encrypted_client(); + let sql = "CREATE TEMPORARY TABLE t (name text)"; + assert_eq!(query(&mut frontend, sql).await, sql); + for sql in [ + "INSERT INTO t (name) VALUES ('Ada')", + "INSERT INTO pg_temp.t (name) VALUES ('Ada')", + "UPDATE T SET name = 'Ada'", + ] { + assert_eq!(query(&mut frontend, sql).await, sql); + } + } + + #[tokio::test] + async fn extended_write_to_a_native_temporary_table_is_forwarded() { + let (mut frontend, _) = encrypted_client(); + for (name, sql) in [ + (&b"create"[..], "CREATE TEMPORARY TABLE t (name text)"), + (&b"insert"[..], "INSERT INTO t (name) VALUES ($1)"), + ] { + let result = frontend + .parse_handler( + operation_id(), + Parse { + statement: Bytes::copy_from_slice(name), + query: Bytes::copy_from_slice(sql.as_bytes()), + parameter_types: Vec::new(), + }, + ) + .await; + assert!(result.is_ok(), "{sql}: {result:?}"); + } + } + + /// A table created by encryption-neutral DDL earlier in the same Query is + /// native, though the schema has not modelled it yet. + #[tokio::test] + async fn write_to_a_table_created_earlier_in_the_query_is_forwarded() { + let (mut frontend, _) = encrypted_client(); + let sql = "CREATE TABLE n (id bigint); INSERT INTO n (id) VALUES (1)"; + let forwarded = + tokio::time::timeout(std::time::Duration::from_secs(5), query(&mut frontend, sql)) + .await + .unwrap(); + assert_eq!(forwarded, sql); + } + + /// The exemption is for writes whose targets are all native tables, not + /// for any write that merely reads one. + #[tokio::test] + async fn native_temporary_table_does_not_excuse_an_encrypted_target() { + let (mut frontend, _) = encrypted_client(); + let forwarded = query( + &mut frontend, + "CREATE TEMPORARY TABLE t (name text); INSERT INTO encrypted (secret) SELECT 'plaintext' FROM t", + ) + .await; + assert!( + forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{forwarded}" + ); + } + + #[tokio::test] + async fn native_temporary_table_does_not_excuse_another_table() { + for sql in [ + "CREATE TEMPORARY TABLE t (name text); INSERT INTO accounts (secret) VALUES ('plaintext')", + "CREATE TEMPORARY TABLE t (name text); INSERT INTO public.t (name) VALUES ('plaintext')", + ] { + let (mut frontend, _) = encrypted_client(); + let forwarded = query(&mut frontend, sql).await; + assert!( + forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{sql}: {forwarded}" + ); + } + } + + #[tokio::test] + async fn refusal_holds_when_mapping_errors_are_enabled() { + let mut config = TandemConfig::for_testing(); + config.disable_mapping_for_testing(); + let development = config.development.as_mut().unwrap(); + development.disable_mapping = false; + development.enable_mapping_errors = true; + let (mut frontend, _) = client(config, true); + assert_refused( + &query( + &mut frontend, + "INSERT INTO accounts (secret) VALUES ('plaintext')", + ) + .await, + "accounts", + ); + } + + #[tokio::test] + async fn write_on_an_unresolvable_table_is_forwarded_without_encrypted_columns() { + let (mut frontend, _) = client(TandemConfig::for_testing(), false); + let sql = "INSERT INTO accounts (secret) VALUES ('plaintext')"; + assert_eq!(query(&mut frontend, sql).await, sql); + } + + /// Read-only statements keep the passthrough. + #[tokio::test] + async fn read_on_an_unresolvable_table_is_forwarded() { + let (mut frontend, _) = encrypted_client(); + let sql = "SELECT * FROM pg_catalog.pg_type WHERE typname = 'text'"; + assert_eq!(query(&mut frontend, sql).await, sql); + } + + /// The loader leaves an encrypted column with no configuration when + /// same-named tables declare it with different domains, so a qualified + /// write is refused rather than encrypted with the other table's config. + #[tokio::test] + async fn write_to_a_column_with_no_config_is_refused() { + let mut schema = eql_mapper::Schema::new("public"); + for table_schema in ["app", "public"] { + let mut table = eql_mapper::Table::in_schema( + sqltk::parser::ast::Ident::new(table_schema), + sqltk::parser::ast::Ident::new("users"), + ); + table.add_column(Arc::new(eql_mapper::Column::eql( + sqltk::parser::ast::Ident::new("email"), + eql_mapper::EqlTraits::none(), + eql_mapper::DomainIdentity::from_domain_name("eql_v3_text").unwrap(), + ))); + schema.add_table(table); + } + // Another encrypted column, so encrypted columns are configured. + let mut encryption = EncryptConfig::new(); + encryption.insert( + crate::Identifier::new("encrypted", "secret"), + cipherstash_client::schema::ColumnConfig::build("secret".to_owned()) + .casts_as(cipherstash_client::schema::ColumnType::Text), + ); + let counter = Counter::default(); + let (sender, _) = tokio::sync::mpsc::unbounded_channel(); + let context = Context::new( + 1, + Arc::new(TandemConfig::for_testing()), + Arc::new(encryption), + Arc::new(schema), + Arc::new(rustls::RootCertStore::empty()), + counter.clone(), + sender, + ); + let mut frontend = Frontend::new(context); + let output = frontend + .intercept( + operation_id(), + FrontendMessage::Query(Bytes::from_static( + b"INSERT INTO app.users (email) VALUES ('plaintext')", + )), + ) + .await; + let forwarded = match output { + Ok(FrontendMiddlewareOutput::Forward(FrontendMessage::Query(query))) + | Ok(FrontendMiddlewareOutput::ForwardThenFlush(FrontendMessage::Query(query))) => { + String::from_utf8_lossy(&query).into_owned() + } + other => panic!("expected a forwarded query, got {other:?}"), + }; + assert!( + forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{forwarded}" + ); + assert_eq!(counter.encrypted.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn write_is_refused_when_only_columns_with_no_config_are_encrypted() { + let mut schema = eql_mapper::Schema::new("public"); + for table_schema in ["app", "public"] { + let mut table = eql_mapper::Table::in_schema( + sqltk::parser::ast::Ident::new(table_schema), + sqltk::parser::ast::Ident::new("users"), + ); + table.add_column(Arc::new(eql_mapper::Column::eql( + sqltk::parser::ast::Ident::new("email"), + eql_mapper::EqlTraits::none(), + eql_mapper::DomainIdentity::from_domain_name("eql_v3_text").unwrap(), + ))); + schema.add_table(table); + } + // The loader dropped the conflicting column's config, leaving none. + let counter = Counter::default(); + let (sender, _) = tokio::sync::mpsc::unbounded_channel(); + let context = Context::new( + 1, + Arc::new(TandemConfig::for_testing()), + Arc::new(EncryptConfig::new()), + Arc::new(schema), + Arc::new(rustls::RootCertStore::empty()), + counter.clone(), + sender, + ); + let mut frontend = Frontend::new(context); + let forwarded = query( + &mut frontend, + "INSERT INTO users (email) VALUES ('plaintext')", + ) + .await; + assert_refused(&forwarded, "users"); + assert_eq!(counter.encrypted.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn schema_qualified_write_is_encrypted() { + let (mut frontend, counter) = encrypted_client(); + let forwarded = query( + &mut frontend, + "INSERT INTO public.encrypted (secret) VALUES ('plaintext')", + ) + .await; + assert_eq!(counter.encrypted.load(Ordering::SeqCst), 1); + assert!(!forwarded.contains("plaintext"), "{forwarded}"); + } + + /// A Query that falls back to passthrough is forwarded whole, as written, + /// so the fallback must not carry another statement that needs encrypting. + #[tokio::test] + async fn passthrough_fallback_does_not_carry_an_encrypted_statement() { + for sql in [ + "SELECT * FROM unknown_view; INSERT INTO encrypted (secret) VALUES ('plaintext')", + "INSERT INTO encrypted (secret) VALUES ('plaintext'); SELECT * FROM unknown_view", + ] { + let (mut frontend, _) = encrypted_client(); + let forwarded = query(&mut frontend, sql).await; + assert!( + forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{sql}: {forwarded}" + ); + } + } + + #[tokio::test] + async fn passthrough_fallback_does_not_carry_a_refused_statement() { + let (mut frontend, _) = encrypted_client(); + assert_refused( + &query( + &mut frontend, + "SELECT * FROM unknown_view; INSERT INTO accounts (secret) VALUES ('plaintext')", + ) + .await, + "accounts", + ); + } + + #[tokio::test] + async fn passthrough_fallback_still_serves_a_query_with_nothing_encrypted() { + let (mut frontend, _) = encrypted_client(); + let sql = "SELECT * FROM unknown_view; SELECT id FROM encrypted"; + assert_eq!(query(&mut frontend, sql).await, sql); + } +} diff --git a/packages/cipherstash-proxy/src/proxy/schema/manager.rs b/packages/cipherstash-proxy/src/proxy/schema/manager.rs index 8dbf8659..9ff56acf 100644 --- a/packages/cipherstash-proxy/src/proxy/schema/manager.rs +++ b/packages/cipherstash-proxy/src/proxy/schema/manager.rs @@ -11,6 +11,7 @@ use arc_swap::ArcSwap; use cipherstash_client::eql::Identifier; use eql_mapper::{Column, Schema, Table}; use sqltk::parser::ast::Ident; +use std::collections::{HashMap, HashSet}; use std::future::Future; use std::sync::{ atomic::{AtomicU64, Ordering}, @@ -385,39 +386,35 @@ fn classify_column( Column::native(ident) } -/// Loads the current structural schema without exposing encryption metadata. -pub async fn load_schema(config: &DatabaseConfig) -> Result { - load_snapshot(config).await.map(|(schema, _)| schema) +/// One row of [`SCHEMA_QUERY`]: a table and its columns. +struct TableRow { + table_schema: String, + table_name: String, + columns: Vec, + column_type_names: Vec>, + column_domain_names: Vec>, } -/// Reads schema and encryption metadata in one repeatable-read transaction. -async fn load_snapshot(config: &DatabaseConfig) -> Result<(Schema, EncryptConfig), Error> { - let client = connect::database(config).await?; - client - .batch_execute("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY") - .await?; - - let tables = client.query(SCHEMA_QUERY, &[]).await?; - +/// Builds the schema and its encryption metadata from the catalog query's rows. +fn build_snapshot(tables: impl IntoIterator) -> (Schema, EncryptConfig) { let mut schema = Schema::new("public"); let mut encrypt_config = EncryptConfig::new(); + // The domain each configured column was loaded with, and the columns whose + // same-named tables disagree on it. + let mut domains: HashMap = HashMap::new(); + let mut conflicting: HashSet = HashSet::new(); + + for row in tables { + let table_name = row.table_name; + let mut table = Table::in_schema( + Ident::with_quote('"', &row.table_schema), + Ident::new(&table_name), + ); - if tables.is_empty() { - warn!(msg = "Database schema contains no tables"); - } - - for table in tables { - let table_name: String = table.get("table_name"); - let columns: Vec = table.get("columns"); - let column_type_names: Vec> = table.get("column_type_names"); - let column_domain_names: Vec> = table.get("column_domain_names"); - - let mut table = Table::new(Ident::new(&table_name)); - - columns + row.columns .iter() - .zip(column_type_names) - .zip(column_domain_names) + .zip(row.column_type_names) + .zip(row.column_domain_names) .for_each(|((col, column_type_name), column_domain_name)| { let column = classify_column( &table_name, @@ -431,11 +428,22 @@ async fn load_snapshot(config: &DatabaseConfig) -> Result<(Schema, EncryptConfig if let Some(domain) = column_domain_name.as_deref() { if let Some(config) = column_config_from_domain(&table_name, col, domain) { let identifier = Identifier::new(table_name.clone(), col.clone()); - // The catalog query orders rows in search-path - // precedence and requires consumers to keep the FIRST - // row for a duplicate name. - if encrypt_config.get_column_config(&identifier).is_none() { - encrypt_config.insert(identifier, config); + // The encrypt config is keyed by the bare table name, + // so same-named tables in two schemas share an entry. + // It is only right for both when they use one domain: + // otherwise a write to one would be encrypted with the + // other's configuration. Leave such a column with no + // configuration, so statements on it are refused. + match domains.get(&identifier) { + None => { + domains.insert(identifier.clone(), domain.to_owned()); + encrypt_config.insert(identifier, config); + } + Some(seen) if seen != domain && conflicting.insert(identifier.clone()) => { + warn!(target: SCHEMA, msg = "Encrypted column is declared with different EQL domains in tables of the same name in different schemas. Statements on this column will be REFUSED, as Proxy keys encryption configuration by the unqualified table name.", table = table_name, column = col, table_schema = row.table_schema); + encrypt_config.remove_column(&table_name, col); + } + Some(_) => {} } } } @@ -444,6 +452,35 @@ async fn load_snapshot(config: &DatabaseConfig) -> Result<(Schema, EncryptConfig schema.add_table(table); } + (schema, encrypt_config) +} + +/// Loads the current structural schema without exposing encryption metadata. +pub async fn load_schema(config: &DatabaseConfig) -> Result { + load_snapshot(config).await.map(|(schema, _)| schema) +} + +/// Reads schema and encryption metadata in one repeatable-read transaction. +async fn load_snapshot(config: &DatabaseConfig) -> Result<(Schema, EncryptConfig), Error> { + let client = connect::database(config).await?; + client + .batch_execute("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY") + .await?; + + let tables = client.query(SCHEMA_QUERY, &[]).await?; + + if tables.is_empty() { + warn!(msg = "Database schema contains no tables"); + } + + let (mut schema, encrypt_config) = build_snapshot(tables.into_iter().map(|table| TableRow { + table_schema: table.get("table_schema"), + table_name: table.get("table_name"), + columns: table.get("columns"), + column_type_names: table.get("column_type_names"), + column_domain_names: table.get("column_domain_names"), + })); + let aggregates = client.query(AGGREGATE_QUERY, &[]).await?; schema.aggregates = aggregates .into_iter() @@ -525,6 +562,82 @@ mod test { ); } + fn users_row(table_schema: &str, email_domain: Option<&str>) -> TableRow { + TableRow { + table_schema: table_schema.to_owned(), + table_name: "users".to_owned(), + columns: vec!["id".to_owned(), "email".to_owned()], + column_type_names: vec![Some("int4".to_owned()), Some("jsonb".to_owned())], + column_domain_names: vec![None, email_domain.map(str::to_owned)], + } + } + + fn object_name(parts: &[&str]) -> sqltk::parser::ast::ObjectName { + sqltk::parser::ast::ObjectName( + parts + .iter() + .map(|part| sqltk::parser::ast::ObjectNamePart::Identifier(Ident::new(*part))) + .collect(), + ) + } + + /// The catalog query returns a row per schema on the search path that has + /// the table. A client can change its own search path, which Proxy does not + /// track, so a bare name used in two schemas stays unresolved and a write + /// through it is refused. A qualified name resolves to its own table. + #[test] + fn a_table_name_in_two_schemas_resolves_only_when_qualified() { + let (schema, encrypt_config) = build_snapshot([ + users_row("app", None), + users_row("public", Some("eql_v3_text_search")), + ]); + + assert!(schema.resolve_table(&object_name(&["users"])).is_err()); + let app = schema + .resolve_table(&object_name(&["app", "users"])) + .unwrap(); + assert_eq!(app.schema, Some(Ident::with_quote('"', "app"))); + let public = schema + .resolve_table(&object_name(&["public", "users"])) + .unwrap(); + assert_eq!(public.schema, Some(Ident::with_quote('"', "public"))); + assert!(encrypt_config + .get_column_config(&Identifier::new("users", "email")) + .is_some()); + } + + /// The encrypt config is keyed by the bare table name, so it can describe + /// a column of two same-named tables only when both use one domain. + /// Otherwise a qualified write would encrypt with the other table's + /// configuration, so neither column gets one and both are refused. + #[test] + fn same_named_encrypted_columns_with_different_domains_get_no_config() { + let (_, encrypt_config) = build_snapshot([ + users_row("app", Some("eql_v3_text_search")), + users_row("public", Some("eql_v3_text")), + ]); + assert!(encrypt_config + .get_column_config(&Identifier::new("users", "email")) + .is_none()); + + let (_, encrypt_config) = build_snapshot([ + users_row("app", Some("eql_v3_text_search")), + users_row("public", Some("eql_v3_text")), + users_row("archive", Some("eql_v3_text_search")), + ]); + assert!(encrypt_config + .get_column_config(&Identifier::new("users", "email")) + .is_none()); + + let (_, encrypt_config) = build_snapshot([ + users_row("app", Some("eql_v3_text_search")), + users_row("public", Some("eql_v3_text_search")), + ]); + assert!(encrypt_config + .get_column_config(&Identifier::new("users", "email")) + .is_some()); + } + #[test] fn an_older_generation_cannot_replace_a_newer_snapshot() { let snapshot = ArcSwap::new(Arc::new(CommittedSchemaSnapshot::new( diff --git a/packages/cipherstash-proxy/src/proxy/schema/middleware.rs b/packages/cipherstash-proxy/src/proxy/schema/middleware.rs index 569dad49..e000c476 100644 --- a/packages/cipherstash-proxy/src/proxy/schema/middleware.rs +++ b/packages/cipherstash-proxy/src/proxy/schema/middleware.rs @@ -72,7 +72,7 @@ use sqltk::parser::ast::{ AlterColumnOperation, AlterTableOperation, ColumnDef, DataType, DropBehavior, Ident, ObjectName, ObjectNamePart, ObjectType, Statement, }; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{ atomic::{AtomicBool, AtomicUsize, Ordering}, Arc, RwLock, @@ -158,6 +158,12 @@ pub struct SchemaMiddleware { local_unmodelled: Arc, transaction_active: Arc, savepoints: Arc>>, + /// Catalog spellings of the native tables this connection has sent to + /// PostgreSQL to create and the schema may not model yet: temporary tables, + /// and tables created earlier in the same batch. Recorded when the + /// statement is sent, so a later statement in the same batch or pipeline + /// sees it too. + native_tables: Arc>>, } impl SchemaMiddleware { @@ -190,6 +196,7 @@ impl SchemaMiddleware { local_unmodelled: Arc::new(AtomicBool::new(false)), transaction_active: Arc::new(AtomicBool::new(false)), savepoints: Arc::new(RwLock::new(Vec::new())), + native_tables: Arc::new(RwLock::new(HashSet::new())), } } @@ -198,6 +205,12 @@ impl SchemaMiddleware { self.resolver.read().unwrap().clone() } + /// Whether the loaded schema has any encrypted column, whether or not the + /// encrypt config names it. + pub fn schema_has_encrypted_columns(&self) -> bool { + self.base.read().unwrap().has_encrypted_columns() + } + /// Returns encryption metadata aligned with the current effective schema. pub fn encrypt_config(&self) -> Arc { self.encrypt_config.read().unwrap().clone() @@ -270,8 +283,47 @@ impl SchemaMiddleware { } } + /// Records the native tables `statements` create. + /// + /// A native table is created by DDL that cannot change encryption: it has + /// no encrypted column and cannot shadow an encrypted table, so a write to + /// it writes no encrypted column even before the schema models it. A + /// temporary table qualifies when [`Self::is_schema_ddl`] ignores it. + pub fn note_native_tables(&self, statements: &[Statement]) { + let mut tables = self.native_tables.write().unwrap(); + for statement in statements { + if let Statement::CreateTable(create) = statement { + let native = if create.temporary { + !self.is_schema_ddl(statement) + } else { + !self.ddl_may_change_encryption(statement) + }; + if native { + tables.insert(postgres_object_name(&create.name)); + } + } + } + } + + /// Whether `name` names a native table this connection created: either + /// bare, or qualified with `pg_temp`. + pub fn is_native_table(&self, name: &ObjectName) -> bool { + let in_temporary_schema = match name.0.as_slice() { + [_] => true, + [ObjectNamePart::Identifier(schema), _] => postgres_identifier(schema) == "pg_temp", + _ => false, + }; + in_temporary_schema + && self + .native_tables + .read() + .unwrap() + .contains(&postgres_object_name(name)) + } + /// Records schema intent for a named prepared statement. pub fn prepare(&self, name: Name, statement: Statement) { + self.note_native_tables(std::slice::from_ref(&statement)); self.prepared.write().unwrap().insert( name, Intent { @@ -319,6 +371,7 @@ impl SchemaMiddleware { /// Records every statement and readiness boundary in a simple-query message. pub fn simple_query(&self, statements: &[Statement]) { + self.note_native_tables(statements); self.transaction_active.store(true, Ordering::Release); let mut executions = self.executions.write().unwrap(); for statement in statements { diff --git a/packages/cipherstash-proxy/src/proxy/sql/select_table_schemas.sql b/packages/cipherstash-proxy/src/proxy/sql/select_table_schemas.sql index d08f207d..c6a2f62a 100644 --- a/packages/cipherstash-proxy/src/proxy/sql/select_table_schemas.sql +++ b/packages/cipherstash-proxy/src/proxy/sql/select_table_schemas.sql @@ -20,11 +20,10 @@ WHERE t.table_type = 'BASE TABLE' -- Only the schemas this connection resolves unqualified names in. -- - -- Both this schema map and the encrypt config are keyed on the bare table - -- name, so without this a same-named table in ANY other schema — another - -- tenant's, a staging copy, a leftover eql_v2 — overwrites the served one, - -- and a column can pick up the wrong domain config or lose its encryption - -- mapping entirely. + -- The schema map records each table's schema, but the encrypt config is + -- keyed on the bare table name, so a same-named table in another schema + -- (another tenant's, a staging copy, a leftover eql_v2) would share or + -- conflict with the served table's column config. -- -- `false` excludes the implicitly-searched pg_catalog, leaving exactly the -- schemas an unqualified `users` could mean. @@ -32,9 +31,9 @@ WHERE GROUP BY t.table_schema, t.table_name ORDER BY - -- Search-path order, so the first row for a name is the one PostgreSQL - -- itself would resolve to. Consumers keep the first and ignore later - -- duplicates. + -- Search-path order, so rows for a name come in the order PostgreSQL + -- would resolve them. Consumers keep every row: a bare name found in two + -- schemas stays ambiguous, and qualified names resolve to their schema. array_position(current_schemas(false)::text[], t.table_schema::text), t.table_name; diff --git a/packages/eql-mapper/src/eql_mapper.rs b/packages/eql-mapper/src/eql_mapper.rs index 031ddbcd..435fb8d7 100644 --- a/packages/eql-mapper/src/eql_mapper.rs +++ b/packages/eql-mapper/src/eql_mapper.rs @@ -8,7 +8,8 @@ use crate::{ use sqltk::parser::ast::{self as ast, Statement}; use sqltk::{Break, NodeKey, Visitable, Visitor}; use std::{ - cell::RefCell, collections::HashMap, marker::PhantomData, ops::ControlFlow, rc::Rc, sync::Arc, + cell::RefCell, collections::HashMap, convert::Infallible, marker::PhantomData, + ops::ControlFlow, rc::Rc, sync::Arc, }; use tracing::{event, span, Level}; @@ -70,6 +71,81 @@ pub fn requires_type_check(statement: &Statement) -> bool { ) } +/// Returns whether the [`Statement`] writes rows: an `INSERT`, `UPDATE`, `DELETE` or `MERGE`, including one nested +/// in a `PREPARE`, an `EXPLAIN` or a data-modifying CTE. +pub fn writes_data(statement: &Statement) -> bool { + struct FindWrite; + + impl<'ast> Visitor<'ast> for FindWrite { + type Error = Infallible; + + fn enter(&mut self, node: &'ast N) -> ControlFlow> { + match node.downcast_ref::() { + Some( + Statement::Insert(_) + | Statement::Update { .. } + | Statement::Delete(_) + | Statement::Merge { .. }, + ) => ControlFlow::Break(Break::Finished), + _ => ControlFlow::Continue(()), + } + } + } + + statement.accept(&mut FindWrite).is_break() +} + +/// Returns the tables the [`Statement`] writes to, found as [`writes_data`] finds its writes, or `None` when a write +/// target is not a plain table name (a table function or a subquery, say). +pub fn write_targets(statement: &Statement) -> Option> { + #[derive(Default)] + struct FindTargets { + names: Vec, + unnamed: bool, + } + + impl FindTargets { + fn relation(&mut self, relation: &ast::TableFactor) { + match relation { + ast::TableFactor::Table { name, .. } => self.names.push(name.clone()), + _ => self.unnamed = true, + } + } + } + + impl<'ast> Visitor<'ast> for FindTargets { + type Error = Infallible; + + fn enter(&mut self, node: &'ast N) -> ControlFlow> { + match node.downcast_ref::() { + Some(Statement::Insert(insert)) => match &insert.table { + ast::TableObject::TableName(name) => self.names.push(name.clone()), + _ => self.unnamed = true, + }, + Some(Statement::Update { table, .. }) => self.relation(&table.relation), + Some(Statement::Delete(delete)) if !delete.tables.is_empty() => { + self.names.extend(delete.tables.iter().cloned()) + } + Some(Statement::Delete(delete)) => match &delete.from { + ast::FromTable::WithFromKeyword(tables) + | ast::FromTable::WithoutKeyword(tables) => { + for table in tables { + self.relation(&table.relation); + } + } + }, + Some(Statement::Merge { table, .. }) => self.relation(table), + _ => {} + } + ControlFlow::Continue(()) + } + } + + let mut targets = FindTargets::default(); + let _ = statement.accept(&mut targets); + (!targets.unnamed).then_some(targets.names) +} + /// The error type returned by various functions in the `eql_mapper` crate. #[derive(Debug, PartialEq, Eq, thiserror::Error)] pub enum EqlMapperError { @@ -119,6 +195,23 @@ impl EqlMapperError { _ => None, } } + + /// Returns the table name when this error means the statement names a + /// table the schema cannot resolve: one that is not loaded, a name that is + /// qualified by another schema, or a name more than one table uses. + /// + /// A caller must not treat such a statement as one that touches nothing + /// encrypted, because the type checker never saw the table's columns. + pub fn as_table_not_found(&self) -> Option<&str> { + match self { + EqlMapperError::Type(err) => err.as_table_not_found(), + EqlMapperError::Scope(err) => err.as_table_not_found(), + EqlMapperError::Import(ImportError::SchemaError(err)) => err.as_table_not_found(), + EqlMapperError::Import(ImportError::ScopeError(err)) => err.as_table_not_found(), + EqlMapperError::Import(ImportError::TypeError(err)) => err.as_table_not_found(), + _ => None, + } + } } /// `EqlMapper` can safely convert a SQL statement into an equivalent statement where all of the plaintext literals have diff --git a/packages/eql-mapper/src/inference/type_error.rs b/packages/eql-mapper/src/inference/type_error.rs index 21e3fe7f..fdbd47be 100644 --- a/packages/eql-mapper/src/inference/type_error.rs +++ b/packages/eql-mapper/src/inference/type_error.rs @@ -139,4 +139,14 @@ impl TypeError { _ => None, } } + + /// Returns the table name when this error means a statement names a table + /// the schema cannot resolve. + pub fn as_table_not_found(&self) -> Option<&str> { + match self { + TypeError::SchemaError(err) => err.as_table_not_found(), + TypeError::ScopeError(err) => err.as_table_not_found(), + _ => None, + } + } } diff --git a/packages/eql-mapper/src/lib.rs b/packages/eql-mapper/src/lib.rs index 4de54147..1443a3bb 100644 --- a/packages/eql-mapper/src/lib.rs +++ b/packages/eql-mapper/src/lib.rs @@ -534,6 +534,181 @@ mod test { } } + /// Asserts the statement fails to type check because it names a table the + /// schema cannot resolve, and returns the name the error reports. + fn table_not_found(schema: Arc, sql: &str) -> String { + let statement = parse(sql); + match type_check(schema, &statement) { + Ok(_) => panic!("statement type checked but names an unknown table: {sql}"), + Err(err) => err + .as_table_not_found() + .unwrap_or_else(|| panic!("failed for another reason ({err}): {sql}")) + .to_owned(), + } + } + + fn users_schema() -> Arc { + resolver(schema! { + tables: { + users: { + id, + email (EQL), + } + } + }) + } + + fn assert_encrypts_users_email(schema: Arc, sql: &str) { + let statement = parse(sql); + match type_check(schema, &statement) { + Ok(typed) => assert!( + typed.literals.contains(&( + EqlTerm::Full(EqlValue::with_canonical_identity( + TableColumn { + table: id("users"), + column: id("email") + }, + EqlTraits::default() + )), + &ast::Value::SingleQuotedString("alice@example.com".into()), + )), + "the literal is not mapped to users.email: {sql}" + ), + Err(err) => panic!("type check failed ({err}): {sql}"), + } + } + + #[test] + fn schema_qualified_table_resolves_to_the_loaded_table() { + assert_encrypts_users_email( + users_schema(), + "INSERT INTO public.users (id, email) VALUES (1, 'alice@example.com')", + ); + assert_encrypts_users_email( + users_schema(), + "UPDATE public.users SET email = 'alice@example.com' WHERE id = 1", + ); + assert_encrypts_users_email( + users_schema(), + "INSERT INTO PUBLIC.USERS (id, email) VALUES (1, 'alice@example.com')", + ); + } + + #[test] + fn table_qualified_by_another_schema_is_not_found() { + assert_eq!( + table_not_found( + users_schema(), + "INSERT INTO archive.users (id, email) VALUES (1, 'alice@example.com')" + ), + "archive.users" + ); + assert_eq!( + table_not_found( + users_schema(), + "INSERT INTO db.public.users (id, email) VALUES (1, 'alice@example.com')" + ), + "db.public.users" + ); + } + + #[test] + fn table_resolves_by_the_schema_it_was_loaded_from() { + let mut schema = Schema::new("public"); + let mut users = crate::Table::in_schema(Ident::new("app"), Ident::new("users")); + users.add_column(Arc::new(crate::Column::native(Ident::new("id")))); + users.add_column(Arc::new(crate::Column::eql( + Ident::new("email"), + EqlTraits::default(), + crate::unifier::DomainIdentity::from_domain_name("eql_v3_text").unwrap(), + ))); + schema.add_table(users); + let schema = resolver(schema); + + assert_encrypts_users_email( + schema.clone(), + "INSERT INTO app.users (id, email) VALUES (1, 'alice@example.com')", + ); + assert_encrypts_users_email( + schema.clone(), + "INSERT INTO users (id, email) VALUES (1, 'alice@example.com')", + ); + // `public.users` is a different table from the loaded `app.users`. + assert_eq!( + table_not_found( + schema, + "INSERT INTO public.users (id, email) VALUES (1, 'alice@example.com')" + ), + "public.users" + ); + } + + #[test] + fn unknown_table_is_reported_wherever_the_statement_names_it() { + for sql in [ + "INSERT INTO accounts (email) VALUES ('alice@example.com')", + "UPDATE accounts SET email = 'alice@example.com'", + "DELETE FROM accounts WHERE email = 'alice@example.com'", + "SELECT * FROM accounts WHERE email = 'alice@example.com'", + "INSERT INTO users (id, email) SELECT id, email FROM accounts", + ] { + assert_eq!(table_not_found(users_schema(), sql), "accounts", "{sql}"); + } + } + + #[test] + fn writes_data_finds_nested_writes() { + for sql in [ + "INSERT INTO users (id) VALUES (1)", + "UPDATE users SET id = 1", + "DELETE FROM users", + "MERGE INTO users USING accounts ON users.id = accounts.id WHEN MATCHED THEN DELETE", + "PREPARE p AS INSERT INTO users (id) VALUES ($1)", + "EXPLAIN ANALYZE DELETE FROM users", + "WITH moved AS (DELETE FROM users RETURNING id) SELECT id FROM moved", + ] { + assert!(crate::writes_data(&parse(sql)), "{sql}"); + } + for sql in [ + "SELECT id FROM users", + "WITH c AS (SELECT id FROM users) SELECT id FROM c", + "EXPLAIN SELECT id FROM users", + ] { + assert!(!crate::writes_data(&parse(sql)), "{sql}"); + } + } + + #[test] + fn write_targets_names_each_table_written() { + let names = |sql: &str| { + crate::write_targets(&parse(sql)).map(|names| { + names + .iter() + .map(|name| name.to_string()) + .collect::>() + }) + }; + assert_eq!( + names("INSERT INTO t (id) SELECT id FROM users"), + Some(vec!["t".to_owned()]) + ); + assert_eq!( + names("UPDATE public.t SET id = 1"), + Some(vec!["public.t".to_owned()]) + ); + assert_eq!( + names("DELETE FROM t USING users"), + Some(vec!["t".to_owned()]) + ); + assert_eq!( + names( + "WITH moved AS (DELETE FROM users RETURNING id) INSERT INTO t SELECT id FROM moved" + ), + Some(vec!["users".to_owned(), "t".to_owned()]) + ); + assert_eq!(names("SELECT id FROM users"), Some(vec![])); + } + #[test] fn insert_with_values_no_explicit_columns() { // init_tracing(); diff --git a/packages/eql-mapper/src/model/schema.rs b/packages/eql-mapper/src/model/schema.rs index a62a0980..390eef95 100644 --- a/packages/eql-mapper/src/model/schema.rs +++ b/packages/eql-mapper/src/model/schema.rs @@ -29,6 +29,10 @@ pub struct Schema { #[derive(Debug, Clone, PartialEq, Eq, Display, Hash)] #[display("Table<{}>", name)] pub struct Table { + /// The database schema the table was loaded from, when it is known. + /// + /// `None` means the table lives in the [`Schema`]'s own namespace. + pub schema: Option, pub name: Ident, pub columns: Vec>, } @@ -109,7 +113,28 @@ pub enum SchemaError { ColumnNotFound(String, String), } +impl SchemaError { + /// Returns the table name when this error means a table could not be resolved. + pub fn as_table_not_found(&self) -> Option<&str> { + match self { + SchemaError::TableNotFound(table) => Some(table), + SchemaError::ColumnNotFound(..) => None, + } + } +} + impl Schema { + /// Whether any column holds encrypted data, including one this build + /// cannot map. + pub fn has_encrypted_columns(&self) -> bool { + self.tables.iter().any(|table| { + table + .columns + .iter() + .any(|column| column.kind != ColumnKind::Native) + }) + } + /// Creates a new named empty schema. pub fn new(name: S) -> Self where @@ -129,19 +154,34 @@ impl Schema { self.tables.push(Arc::new(table)); } - /// Resolves a table by `Ident`, which takes into account the SQL rules + /// Resolves a table by name, which takes into account the SQL rules /// of quoted and new identifier matching. + /// + /// A one-part name matches a table of that name in any loaded schema. A + /// two-part `schema.table` name matches only a table loaded from that + /// schema. A name more than one table matches is not resolved. pub fn resolve_table(&self, name: &ObjectName) -> Result, SchemaError> { - if name.0.len() == 1 { - let ObjectNamePart::Identifier(name) = name.0.last().unwrap(); - let mut haystack = self.tables.iter(); - haystack - .find_unique(&|table| IdentCase::from(&table.name) == IdentCase::from(name)) - .cloned() - .map_err(|_| SchemaError::TableNotFound(name.to_string())) - } else { - Err(SchemaError::TableNotFound(format!("{name}"))) - } + let not_found = || SchemaError::TableNotFound(name.to_string()); + let parts: Vec<&Ident> = name + .0 + .iter() + .map(|ObjectNamePart::Identifier(ident)| ident) + .collect(); + + let (qualifier, table_name) = match parts.as_slice() { + [table_name] => (None, *table_name), + [qualifier, table_name] => (Some(*qualifier), *table_name), + _ => return Err(not_found()), + }; + + let mut haystack = self.tables.iter(); + haystack + .find_unique(&|table| { + IdentCase::from(&table.name) == IdentCase::from(table_name) + && qualifier.is_none_or(|qualifier| table.is_in_schema(&self.name, qualifier)) + }) + .cloned() + .map_err(|_| not_found()) } pub fn resolve_table_columns( @@ -165,30 +205,17 @@ impl Schema { table_name: &ObjectName, column_name: &Ident, ) -> Result { - if table_name.0.len() == 1 { - let ObjectNamePart::Identifier(table_name) = table_name.0.last().unwrap(); - let mut haystack = self.tables.iter(); - match haystack - .find_unique(&|table| IdentCase::from(&table.name) == IdentCase::from(table_name)) - { - Ok(table) => match table.get_column(column_name) { - Ok(column) => Ok(SchemaTableColumn { - table: table.name.clone(), - column: column.name.clone(), - kind: column.kind.clone(), - }), - Err(_) => Err(SchemaError::ColumnNotFound( - table_name.to_string(), - column_name.to_string(), - )), - }, - Err(_) => Err(SchemaError::TableNotFound(table_name.to_string())), - } - } else { - Err(SchemaError::ColumnNotFound( - format!("{table_name}"), - format!("{column_name}"), - )) + let table = self.resolve_table(table_name)?; + match table.get_column(column_name) { + Ok(column) => Ok(SchemaTableColumn { + table: table.name.clone(), + column: column.name.clone(), + kind: column.kind.clone(), + }), + Err(_) => Err(SchemaError::ColumnNotFound( + table_name.to_string(), + column_name.to_string(), + )), } } } @@ -197,11 +224,27 @@ impl Table { /// Create a new named table with no columns. pub fn new(name: Ident) -> Self { Self { + schema: None, name, columns: Vec::with_capacity(16), } } + /// Create a new named table with no columns, in the named database schema. + pub fn in_schema(schema: Ident, name: Ident) -> Self { + Self { + schema: Some(schema), + ..Self::new(name) + } + } + + /// Whether `qualifier` names the database schema this table belongs to. + /// A table with no schema of its own belongs to `default_schema`. + fn is_in_schema(&self, default_schema: &Ident, qualifier: &Ident) -> bool { + let schema = self.schema.as_ref().unwrap_or(default_schema); + IdentCase::from(schema) == IdentCase::from(qualifier) + } + /// Adds a column to the table. pub fn add_column(&mut self, column: Arc) -> Arc { self.columns.push(column); diff --git a/packages/eql-mapper/src/model/schema_delta.rs b/packages/eql-mapper/src/model/schema_delta.rs index 8f6e3068..9008394f 100644 --- a/packages/eql-mapper/src/model/schema_delta.rs +++ b/packages/eql-mapper/src/model/schema_delta.rs @@ -75,7 +75,8 @@ impl SchemaWithEdits { fn get_overlay_mut(&mut self, table_name: &ObjectName) -> &mut Overlay { let key = self.canonical_key(table_name); let overlay = { - let schema_table = self.schema.resolve_table(&key); + // As in `resolve_table`, look the table up by the caller's name. + let schema_table = self.schema.resolve_table(table_name); match schema_table { Ok(schema_table) => Overlay::Table(OverlayTable::from(&*schema_table)), Err(_) => Overlay::Table(OverlayTable::new(key.clone())), @@ -92,7 +93,10 @@ impl SchemaWithEdits { Overlay::Dropped => Err(SchemaError::TableNotFound(name.to_string())), Overlay::Table(overlay_table) => Ok(Arc::new(overlay_table.into())), }, - None => self.schema.resolve_table(&key), + // The caller's name, not the overlay key: the key drops a + // qualifier naming this schema, which would let `public.users` + // resolve to a `users` loaded from another schema. + None => self.schema.resolve_table(name), } } @@ -217,7 +221,15 @@ impl OverlayTable { impl From<&Table> for OverlayTable { fn from(value: &Table) -> Self { Self { - name: ObjectName(vec![ObjectNamePart::Identifier(value.name.clone())]), + name: ObjectName( + value + .schema + .iter() + .chain([&value.name]) + .cloned() + .map(ObjectNamePart::Identifier) + .collect(), + ), columns: value.columns.iter().map(|col| (**col).clone()).collect(), } } @@ -227,8 +239,13 @@ impl From<&OverlayTable> for Table { fn from(value: &OverlayTable) -> Self { let ObjectName(parts) = &value.name; let ObjectNamePart::Identifier(ident) = parts.last().unwrap(); + let schema = match parts.as_slice() { + [ObjectNamePart::Identifier(schema), _] => Some(schema.clone()), + _ => None, + }; Self { + schema, name: ident.clone(), columns: value.columns.iter().cloned().map(Arc::new).collect(), } @@ -459,6 +476,22 @@ mod test { .is_ok()); } + #[test] + fn qualifier_naming_this_schema_does_not_resolve_a_table_loaded_from_another() { + let mut schema = crate::Schema::new("public"); + schema.add_table(crate::Table::in_schema( + Ident::new("app"), + Ident::new("users"), + )); + let resolver = Arc::new(TableResolver::new_editable(Arc::new(schema))); + + assert!(resolver.resolve_table(&qualified("app", "users")).is_ok()); + assert_eq!( + resolver.resolve_table(&qualified("public", "users")), + Err(SchemaError::TableNotFound("public.users".into())) + ); + } + #[test] fn bare_create_resolves_by_schema_qualified_name() { let schema = Arc::new(schema! { tables: {} }); diff --git a/packages/eql-mapper/src/scope_tracker.rs b/packages/eql-mapper/src/scope_tracker.rs index b15d77e7..bc4ffe5e 100644 --- a/packages/eql-mapper/src/scope_tracker.rs +++ b/packages/eql-mapper/src/scope_tracker.rs @@ -348,6 +348,16 @@ pub enum ScopeError { NoCurrentScope, } +impl ScopeError { + /// See [`TypeError::as_table_not_found`]. + pub fn as_table_not_found(&self) -> Option<&str> { + match self { + ScopeError::TypeError(err) => err.as_table_not_found(), + _ => None, + } + } +} + impl<'ast> Visitor<'ast> for ScopeTracker<'ast> { type Error = ScopeError;