From 109eca9d2eacba43e65a5b73aefe2f354f8c67b7 Mon Sep 17 00:00:00 2001 From: Toby Hede Date: Fri, 23 Oct 2026 02:21:49 +0000 Subject: [PATCH] fix(proxy): route each statement through one refuse-by-default decision Frontend::route is now the only place that decides what happens to a statement: map it, forward it for a named reason, or refuse it. Only a coverage gap (a statement EQL Mapper could not type check, or an internal error) may fall back to forwarding the statement as written, and only for a read while mapping errors are off. Any other error, including one added later, refuses. While any column is encrypted, a write never falls back. So that writes to native tables still type check instead of being refused, Proxy's schema now covers them: - A connection's native temporary tables live in a temporary layer that resolves before permanent tables. A table is added once PostgreSQL confirms its CREATE, including CREATE TABLE pg_temp.name, and removed by DROP, DISCARD, ON COMMIT DROP, rollback and rollback to a savepoint. A bare name that also names a permanent table is ambiguous and refused. - In a simple Query, each statement is type checked against the schema plus the encryption-neutral DDL before it in the same Query. CREATE TABLE IF NOT EXISTS is not modelled, since the table may already exist with encrypted columns Proxy has not loaded. Also refuses a write nested in a statement Proxy does not type check, such as COPY (INSERT ... RETURNING ...) TO STDOUT; an operation an encrypted column's EQL type does not support, which would otherwise reach PostgreSQL with a plaintext operand or compare ciphertext; and counts encrypted columns in the loaded schema even when the encrypt config has none for them. --- CHANGELOG.md | 6 + docs/errors.md | 70 +- .../src/extended_protocol_error_messages.rs | 4 +- packages/cipherstash-proxy/CONTEXT.md | 26 +- packages/cipherstash-proxy/src/error.rs | 35 +- .../src/postgresql/context/mod.rs | 28 +- .../src/postgresql/middleware/frontend.rs | 728 ++++++++++++------ .../src/proxy/schema/middleware.rs | 704 +++++++++++++++-- packages/eql-mapper/src/eql_mapper.rs | 80 +- .../eql-mapper/src/inference/type_error.rs | 23 +- packages/eql-mapper/src/lib.rs | 43 +- packages/eql-mapper/src/model/schema.rs | 45 +- .../eql-mapper/src/model/table_resolver.rs | 171 +++- packages/eql-mapper/src/scope_tracker.rs | 16 +- tests/python/tests/test_error_messages.py | 4 +- 15 files changed, 1524 insertions(+), 459 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 65a4e470c..96be8fab4 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 +- **Unknown column error message**: the type-checking error for a column a table does not have now names the column and the table the right way round, for example `Column: email not found for table: users`. It previously swapped them. + - **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. @@ -75,6 +77,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). 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. +- **Writes that fail type checking**: when encrypted columns are configured, including legacy EQL v2 columns, Proxy now refuses a statement that writes data and cannot be type checked, such as `MERGE` or SQL `PREPARE` of an `INSERT`, whether or not `CS_DEVELOPMENT__ENABLE_MAPPING_ERRORS` is set. Previously such a statement was forwarded as written, so its values could reach PostgreSQL in plaintext. This applies to writes to any table while any column is encrypted, including bulk `UPDATE ... FROM (VALUES ...)` and `INSERT ... SELECT ... FROM unnest(...)`, which the type checker does not support yet, and to a write inside `COPY (...) TO`. Read-only statements that cannot be type checked are still forwarded, except one that uses an operation an encrypted column's EQL type does not support, which is now refused. That covers a comparison such as `<` on a column with no ordering index, whose operand would otherwise reach PostgreSQL as plaintext, and `ORDER BY`, `GROUP BY`, `DISTINCT` or `UNION` over an encrypted column without the ordering or equality index they need, which PostgreSQL would otherwise evaluate over the ciphertext and return wrong results without an error. See [Unsupported encrypted operation](docs/errors.md#mapping-unsupported-encrypted-operation). Proxy now tracks each connection's native temporary tables, and checks each statement in a simple query against the schema as the encryption-neutral DDL earlier in that query leaves it, so writes to those tables type check like any other. A write to a bare name that names both a temporary and a permanent table is refused as ambiguous; qualify it, as in `pg_temp.name`. See [Ambiguous table](docs/errors.md#mapping-ambiguous-table). Every statement in a simple query is now checked, including those after one that cannot be type checked. See [Statement could not be type checked](docs/errors.md#mapping-statement-could-not-be-type-checked). + + Compatibility note: before upgrading, check that applications do not send these writes through Proxy while encrypted columns are configured: `MERGE`; SQL `PREPARE` of a write; a write inside `COPY (...) TO`; bulk `UPDATE ... FROM (VALUES ...)`; `INSERT ... SELECT ... FROM unnest(...)`; and other writes Proxy cannot type check, including writes to tables with no encrypted columns. Reads that use an operation an encrypted column does not support, including `ORDER BY`, `GROUP BY`, `DISTINCT` or `UNION` over a column without the index they need, are now refused too. + - **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 b7c6023f0..c2379856b 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -18,6 +18,8 @@ - [Unmodelled DDL](#mapping-unmodelled-ddl) - [Unmappable encrypted column](#mapping-unmappable-encrypted-column) - [Unresolvable table](#mapping-unresolvable-table) + - [Ambiguous table](#mapping-ambiguous-table) + - [Unsupported encrypted operation](#mapping-unsupported-encrypted-operation) - [Unknown prepared statement](#mapping-unknown-prepared-statement) - [Internal Error](#mapping-internal-error) @@ -245,7 +247,18 @@ CipherStash Proxy checks SQL statements against the database schema to transpare The behaviour of Proxy depends on the `mapping_errors_enabled` configuration. -When `mapping_errors_enabled` is `false` (the default), then type check errors are logged, and the statement is passed through to the database. +When `mapping_errors_enabled` is `false` (the default), then type check errors are logged, and a read-only statement is passed through to the database. +A statement that writes data (`INSERT`, `UPDATE`, `DELETE` or `MERGE`, including one inside `PREPARE`, `EXPLAIN`, `COPY (...) TO` or a `WITH` clause) is refused instead whenever any column is encrypted, because Proxy cannot tell whether it writes plaintext to an encrypted column. +This applies to every table, including tables with no encrypted columns, because Proxy cannot tell which columns a write it cannot type check reaches. +For example, these statements are refused while any column is encrypted: + +- `UPDATE users SET name = v.name FROM (VALUES (1, 'Ada')) AS v(id, name) WHERE users.id = v.id` +- `INSERT INTO users (id) SELECT x FROM unnest(ARRAY[1, 2]) AS x` +- `INSERT INTO names SELECT relname FROM pg_catalog.pg_class` + +Proxy type checks writes to the connection's native temporary tables, and to tables that encryption-neutral DDL earlier in the same simple query creates or alters, so those are not refused. +`CREATE TABLE IF NOT EXISTS` is the exception: the table may already exist with encrypted columns that Proxy has not loaded, so a write after it in the same simple query is refused unless an earlier statement created the table. +In a simple query, a statement is also refused when another statement in the same query needs encrypting. When `mapping_errors_enabled` is `true`, then type check errors are raised, and statement execution halts. @@ -254,7 +267,7 @@ Configure this setting with the environment variable `CS_DEVELOPMENT__ENABLE_MAP In our experience, most production systems have a relatively small number of columns that require protection. As SQL is large and complex, instead of blocking statements with type check errors that are false negatives, the default behaviour of Proxy is to allow the statement. -However, this does mean it is possible that a statement that references encrypted columns cannot be type-checked, and it will be passed through to the database. +However, this does mean it is possible that a read-only statement that references encrypted columns cannot be type-checked, and it will be passed through to the database. When a statement is passed through to the database, the database's column constraints (provided by EQL) will catch the statement, and return a PostgreSQL error. Example constraint error: @@ -421,7 +434,7 @@ Statement refers to table 'users', which is not in the database schema Proxy has ### Notes -The refusal applies to `INSERT`, `UPDATE`, `DELETE` and `MERGE`, including one inside `PREPARE`, `EXPLAIN` or a `WITH` clause, when any column is encrypted. +The refusal applies to `INSERT`, `UPDATE`, `DELETE` and `MERGE`, including one inside `PREPARE`, `EXPLAIN`, `COPY (...) TO` 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. @@ -443,6 +456,57 @@ If the table is in a schema outside the search path, add that schema to the sear If the table was created on another connection, retry once Proxy has reloaded the database schema. + + + +## Ambiguous table + +A statement that writes data refers to a table by a bare name that the connection uses for both a temporary table and a permanent table. +PostgreSQL resolves the name to the temporary table, but if that table has been dropped somewhere CipherStash Proxy cannot see, such as inside a function or a `DO` block, the name means the permanent table. +Proxy cannot tell which table the statement writes to, so it refuses the statement rather than forward its values unencrypted. + +### Error message + +``` +Statement refers to table 'notes', which names both a temporary table and a permanent table, so Proxy cannot tell which one it writes to. Qualify the name, as in 'pg_temp.notes'. +``` + +### Notes + +The refusal applies to the same statements as [Unresolvable table](#mapping-unresolvable-table), and only when any column is encrypted. +Read-only statements are not refused. + +### How to fix + +Qualify the table name: `pg_temp.notes` for the temporary table, or the permanent table's schema, as in `public.notes`. +Alternatively, give the temporary table a name no permanent table uses. + +Proxy assumes PostgreSQL's default search order, in which `pg_temp` comes first. +It does not track `SET search_path`, so a client that puts `pg_temp` later in its search path should qualify table names. + + + + + +## Unsupported encrypted operation + +A statement uses an operation on an encrypted column that the column's EQL type does not support, such as `<` or `ORDER BY` on a column with no ordering index, or `LIKE` on a column with no match index. +CipherStash Proxy cannot encrypt the operand for that operation, so it refuses the statement rather than send the operand to the database as plaintext. + +### Error message + +``` +Statement uses an operation that an encrypted column's EQL type does not support: Type `...` does not satisfy bounds `Ord`. +``` + +### Notes + +The refusal applies to reads as well as writes, and whether or not `CS_DEVELOPMENT__ENABLE_MAPPING_ERRORS` is set. + +### How to fix + +Change the statement to use only operations the column supports, or change the column's EQL domain type to one that supports the operation. + diff --git a/packages/cipherstash-proxy-integration/src/extended_protocol_error_messages.rs b/packages/cipherstash-proxy-integration/src/extended_protocol_error_messages.rs index 1e135d4a8..c6e5ec888 100644 --- a/packages/cipherstash-proxy-integration/src/extended_protocol_error_messages.rs +++ b/packages/cipherstash-proxy-integration/src/extended_protocol_error_messages.rs @@ -48,10 +48,12 @@ mod tests { let sql = "INSERT INTO encrypted (id, encrypted_unconfigured) VALUES ($1, $2)"; let result = client.query(sql, &[&id, &encrypted_text]).await; + // EQL Mapper cannot type check a write to a column it does not know, so + // Proxy refuses it rather than forward the values as written. assert_db_error( result, "ERROR", - "column \"encrypted_unconfigured\" of relation \"encrypted\" does not exist", + "Statement could not be type checked: Column: encrypted_unconfigured not found for table: encrypted. For help visit https://github.com/cipherstash/proxy/blob/main/docs/errors.md#mapping-statement-could-not-be-type-checked", ); } diff --git a/packages/cipherstash-proxy/CONTEXT.md b/packages/cipherstash-proxy/CONTEXT.md index 2f678bf6b..41d33c03b 100644 --- a/packages/cipherstash-proxy/CONTEXT.md +++ b/packages/cipherstash-proxy/CONTEXT.md @@ -54,14 +54,28 @@ _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. Four independent conditions produce -it and they are not interchangeable: no encrypt config is loaded, mapping is disabled by -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. +Traffic forwarded without encryption or rewriting. Mapping disabled by configuration +forwards the whole connection. Otherwise each statement's **route** decides, and a +statement is forwarded only for a named reason: `SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING` +was issued on this connection, EQL Mapper does not type check its kind, it touches no +encrypted column, or the **type-check fallback**. These are not interchangeable. Name the condition rather than saying "passthrough" alone. +**Route**: +What Proxy does with one statement from a Query or Parse: map it, forward it for a named +reason, or refuse it. Refusing is the default. A failed type check falls back to +forwarding only for a coverage gap (the statement could not be type checked, or an +internal mapper error), only while mapping errors are off, and never in a simple Query with another statement to map. +When encrypted columns are configured, a write never falls back. +Each statement in a simple Query is type checked against a **batch view**: the effective schema plus the encryption-neutral DDL of the statements before it. + +**Connection temporary tables**: +The native temporary tables one connection has created, which the type checker resolves before permanent tables, as `pg_temp` is searched first. +A table is added when PostgreSQL confirms its `CREATE`, and removed by `DROP`, `DISCARD`, `ON COMMIT DROP` at commit, or rollback of the transaction or savepoint that created it. +A bare name that also names a permanent table is ambiguous and does not resolve, because a stale record would otherwise hide the permanent table. +A write that uses it is refused with its own error. +Proxy does not track `SET search_path`, so it assumes PostgreSQL's default of searching `pg_temp` first; a client that moves `pg_temp` later in its search path can make a bare name mean a permanent table Proxy has not loaded. + ### Encryption **Column**: diff --git a/packages/cipherstash-proxy/src/error.rs b/packages/cipherstash-proxy/src/error.rs index 532cbc5b1..4921ab724 100644 --- a/packages/cipherstash-proxy/src/error.rs +++ b/packages/cipherstash-proxy/src/error.rs @@ -59,30 +59,17 @@ pub enum Error { } impl Error { - /// Whether this error must be returned to the client even when the proxy is - /// configured to fall back to forwarding statements it could not map. - /// - /// That fallback exists for mapper *coverage gaps*: a statement Proxy does - /// not understand is passed through unchanged, on the reasoning that it - /// probably touches nothing encrypted. Some errors are not coverage gaps but - /// deliberate refusals, where forwarding the statement is the very harm the - /// error was raised to prevent. Those are listed here and are never - /// downgraded to a passthrough. - pub fn must_fail_closed(&self) -> bool { + /// Whether this error is a mapper *coverage gap*: EQL Mapper could not type + /// check the statement, which is not evidence that it touches anything + /// encrypted. Only a coverage gap may fall back to forwarding the + /// statement as written. Every other error is a refusal, so an error + /// added later fails closed without being listed anywhere. + pub fn is_coverage_gap(&self) -> bool { matches!( 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. Nor may a - // write to a table Proxy cannot resolve be forwarded. Error::Mapping( - MappingError::UnmappableEncryptedColumn { .. } - | MappingError::UnresolvableTable { .. } - | MappingError::DependentStatementAfterDdl - | MappingError::DependentStatementAfterSettings - | MappingError::PreparedSettingsChanged - | MappingError::UnmodelledDdl - ) | Error::Encrypt(EncryptError::InvalidInboundEqlPayload) + MappingError::StatementCouldNotBeTypeChecked(_) | MappingError::Internal(_) + ) ) } } @@ -158,6 +145,12 @@ pub enum MappingError { #[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("Statement refers to table '{table}', which names both a temporary table and a permanent table, so Proxy cannot tell which one it writes to. Qualify the name, as in 'pg_temp.{table}'. For help visit {}#mapping-ambiguous-table", ERROR_DOC_BASE_URL)] + AmbiguousTable { table: String }, + + #[error("Statement uses an operation that an encrypted column's EQL type does not support: {}. For help visit {}#mapping-unsupported-encrypted-operation", _0, ERROR_DOC_BASE_URL)] + UnsupportedEncryptedOperation(String), + #[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 c41a00e6e..6dd09870d 100644 --- a/packages/cipherstash-proxy/src/postgresql/context/mod.rs +++ b/packages/cipherstash-proxy/src/postgresql/context/mod.rs @@ -1346,9 +1346,10 @@ where .map_err(|_| crate::error::ContextError::ProtocolStateUnavailable.into()) } - /// Returns the resolver for this connection's effective schema snapshot. + /// Returns the resolver for this connection's effective schema snapshot, + /// with its native temporary tables. pub fn get_table_resolver(&self) -> Arc { - self.schema_middleware.resolver() + self.schema_middleware.effective_resolver() } /// Records schema intent for a parsed prepared statement. @@ -1732,21 +1733,14 @@ 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)) - }) + /// The resolver to type check each statement of a simple-query batch with, + /// which sees the encryption-neutral schema changes of the statements + /// before it. + pub fn batch_table_resolvers( + &self, + statements: &[sqltk::parser::ast::Statement], + ) -> Vec> { + self.schema_middleware.batch_resolvers(statements) } /// Whether any column is encrypted. A column can be encrypted without diff --git a/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs b/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs index 88933e44c..f604358b3 100644 --- a/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs +++ b/packages/cipherstash-proxy/src/postgresql/middleware/frontend.rs @@ -27,7 +27,10 @@ use crate::prometheus::{ use crate::proxy::EncryptionService; use crate::{EqlOutput, EqlQueryPayload}; use cipherstash_client::encryption::Plaintext; -use eql_mapper::{self, EqlMapperError, EqlTermVariant, JsonSelectorSegment, TypeCheckedStatement}; +use eql_mapper::{ + self, EqlMapperError, EqlTermVariant, JsonSelectorSegment, TableResolver, TypeCheckedStatement, + UnresolvedTable, +}; use metrics::{counter, histogram}; use pg_proto::{Describe, Execute, FrontendMessage, FrontendMiddlewareOutput, Parse}; use serde::Serialize; @@ -38,6 +41,66 @@ use std::sync::Arc; use std::time::Instant; use tracing::{debug, warn}; +/// What Proxy does with one statement, decided by [`Frontend::route`]. A +/// refusal is the `Err` of that decision rather than a variant, so it cannot be +/// matched past. +// One short-lived Route per statement in a message, so its size doesn't matter. +#[allow(clippy::large_enum_variant)] +enum Route<'a> { + /// Encrypt and rewrite the statement. + Map { + typed: TypeCheckedStatement<'a>, + statement: Statement, + }, + /// Forward the statement as the client wrote it. + Forward(Forward), +} + +impl Route<'_> { + fn is_map(&self) -> bool { + matches!(self, Route::Map { .. }) + } + + fn is_fallback(&self) -> bool { + matches!(self, Route::Forward(Forward::Fallback(_))) + } +} + +/// Why a statement is forwarded as written. Only `NothingEncrypted` is backed +/// by a type check; the others are policy, and each is named so it can be +/// found and tested. +enum Forward { + /// `SET CIPHERSTASH.UNSAFE_DISABLE_MAPPING` on this connection. + MappingDisabled, + /// A statement kind EQL Mapper does not type check. + NotTypeChecked, + /// Type checked, and it touches no encrypted column. + NothingEncrypted, + /// The type-check fallback, carrying the error it overrode. + Fallback(Error), +} + +impl Forward { + fn record(&self) { + match self { + Forward::MappingDisabled => { + warn!(msg = "Encrypted statement mapping is not enabled"); + counter!(STATEMENTS_PASSTHROUGH_MAPPING_DISABLED_TOTAL).increment(1); + counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); + } + Forward::NotTypeChecked => { + counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); + } + Forward::NothingEncrypted => { + debug!(target: MAPPER, msg = "Passthrough Statement"); + counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); + } + // `type_check` already logged and counted the error. + Forward::Fallback(_) => {} + } + } +} + /// The PostgreSQL proxy frontend that handles client-to-server message processing. /// /// The Frontend intercepts messages from PostgreSQL clients, analyzes SQL statements for @@ -369,9 +432,6 @@ 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) @@ -386,62 +446,40 @@ impl Frontend { statements = parsed_statements.len(), ); + // Route every statement before acting on any, so the batch rule below + // sees them all and no statement goes unchecked behind a fallback. + let mut routes = parsed_statements + .iter() + .zip(self.context.batch_table_resolvers(&parsed_statements)) + .map(|(statement, resolver)| self.route(statement, vec![], &resolver)) + .collect::, _>>()?; + + // A fallback forwards the whole Query as the client wrote it, so it + // must not carry a statement that needs mapping: that statement's + // values would reach the database unencrypted. + if routes.iter().any(Route::is_map) { + if let Some(index) = routes.iter().position(Route::is_fallback) { + let Route::Forward(Forward::Fallback(err)) = routes.swap_remove(index) else { + unreachable!("position found a fallback"); + }; + return Err(err); + } + } + let mut portal = Portal::passthrough(Some(session_id)); let mut encrypted = false; - let mut mapped = false; let mut parse_duration_recorded = false; - 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); - counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); - forwarded_statements.push(statement.clone()); - continue; - } - - if !eql_mapper::requires_type_check(statement) { - counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); - forwarded_statements.push(statement.clone()); - continue; - } - - let typed_statement = match self.type_check(statement) { - Ok(ts) => ts, - Err(err) => { - // `must_fail_closed` errors are refusals: passing the - // original statement to the database is the harm they exist - // to prevent, so they ignore the passthrough fallback. This - // holds today, independently of CIP-3680 removing that - // 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, - session_id, - Portal::passthrough(Some(session_id)), - &parsed_statements, - )?; - return Ok(None); - }; - } - }; - - match self.to_encryptable_statement(&typed_statement, vec![])? { - Some(statement) => { + for (statement, route) in parsed_statements.iter().zip(routes) { + match route { + Route::Map { + typed: typed_statement, + statement: encryptable, + } => { debug!(target: MAPPER, client_id = self.context.client_id, msg = "Encryptable Statement", ); - mapped = true; let mut transformed = false; if typed_statement.requires_transform() { @@ -456,7 +494,7 @@ impl Frontend { .encrypt_literals( session_id, &typed_statement, - &statement.literal_columns, + &encryptable.literal_columns, ) .await?; @@ -483,20 +521,16 @@ impl Frontend { counter!(STATEMENTS_ENCRYPTED_TOTAL).increment(1); // Set Encrypted portal and mark as mapped - portal = Portal::encrypted(Arc::new(statement), Some(session_id)); + portal = Portal::encrypted(Arc::new(encryptable), Some(session_id)); self.context.update_statement_metadata(session_id, |m| { m.encrypted = true; })?; } - None => { - debug!(target: MAPPER, - client_id = self.context.client_id, - msg = "Passthrough Statement" - ); - counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); + Route::Forward(reason) => { + reason.record(); forwarded_statements.push(statement.clone()); } - }; + } } // Record parse/typecheck duration (if not already recorded before encryption) @@ -783,34 +817,6 @@ impl Frontend { .prepare_parse(operation, &statement, &statement_text) .await?; - if self.context.unsafe_disable_mapping()? { - warn!(msg = "Encrypted statement mapping is not enabled"); - counter!(STATEMENTS_PASSTHROUGH_MAPPING_DISABLED_TOTAL).increment(1); - counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); - self.context.finish_parse(operation, None)?; - return Ok(None); - } - - if !eql_mapper::requires_type_check(&statement) { - counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); - self.context.finish_parse(operation, None)?; - return Ok(None); - } - - let typed_statement = match self.type_check(&statement) { - Ok(ts) => ts, - Err(err) => { - // See the matching comment on the simple-protocol path: a - // refusal is never downgraded to a passthrough. - if self.context.mapping_errors_enabled() || err.must_fail_closed() { - return Err(err); - } else { - self.context.finish_parse(operation, None)?; - return Ok(None); - }; - } - }; - // Capture the parse message param_types // These override the underlying column type let client_param_types = message.parameter_types.clone(); @@ -818,8 +824,11 @@ impl Frontend { let mut parse_duration_recorded = false; - match self.to_encryptable_statement(&typed_statement, param_types)? { - Some(mut statement) => { + match self.route(&statement, param_types, &self.context.get_table_resolver())? { + Route::Map { + typed: typed_statement, + mut statement, + } => { if typed_statement.requires_transform() { // Record parse duration before encryption work starts self.context @@ -859,13 +868,9 @@ impl Frontend { rewrite_parse_param_types(&client_param_types, &statement.output_params); self.context.finish_parse(operation, Some(statement))?; } - _ => { + Route::Forward(reason) => { + reason.record(); self.context.finish_parse(operation, None)?; - debug!(target: MAPPER, - client_id = self.context.client_id, - msg = "Passthrough Parse" - ); - counter!(STATEMENTS_PASSTHROUGH_TOTAL).increment(1); } } @@ -1228,102 +1233,148 @@ 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; + /// Decides what Proxy does with one statement: map it, forward it for a + /// named reason, or refuse it (`Err`). This is the only place that decides, + /// and refusing is the default: a failed type check is forwarded only when + /// [`Self::may_fall_back`] allows it. `resolver` is the schema to type check + /// the statement against. + fn route<'a>( + &self, + statement: &'a ast::Statement, + param_types: Vec, + resolver: &Arc, + ) -> Result, Error> { + if self.context.unsafe_disable_mapping()? { + return Ok(Route::Forward(Forward::MappingDisabled)); + } + if !eql_mapper::requires_type_check(statement) { + // A write nested in a statement the type checker skips, such as + // `COPY (INSERT ... RETURNING ...) TO STDOUT`, is never mapped. + if self.context.has_encrypted_columns() && eql_mapper::writes_data(statement) { + return Err(MappingError::StatementCouldNotBeTypeChecked( + "it writes data inside a statement Proxy does not type check".to_owned(), + ) + .into()); } - match self.type_check(statement) { - Ok(typed_statement) => { - if self - .to_encryptable_statement(&typed_statement, vec![])? - .is_some() - { - return Ok(true); - } + return Ok(Route::Forward(Forward::NotTypeChecked)); + } + match self.type_check(statement, resolver) { + Ok(typed) => match self.to_encryptable_statement(&typed, param_types)? { + Some(encryptable) => Ok(Route::Map { + typed, + statement: encryptable, + }), + None => Ok(Route::Forward(Forward::NothingEncrypted)), + }, + Err(err) => { + // The one place the write policy is checked: it decides both + // the error a refused write gets and whether it may fall back. + let writes_encrypted = + self.context.has_encrypted_columns() && eql_mapper::writes_data(statement); + let err = self.mapping_error(err, writes_encrypted); + if self.may_fall_back(&err, writes_encrypted) { + Ok(Route::Forward(Forward::Fallback(err))) + } else { + Err(err) } - Err(err) if err.must_fail_closed() => return Err(err), - Err(_) => {} } } - Ok(false) + } + + /// The type-check fallback: a read that hit a coverage gap, only while + /// mapping errors are off. A write never falls back while encrypted columns + /// are configured, since it could put plaintext in an encrypted column. + fn may_fall_back(&self, err: &Error, writes_encrypted: bool) -> bool { + !self.context.mapping_errors_enabled() && !writes_encrypted && err.is_coverage_gap() } fn type_check<'a>( &self, statement: &'a ast::Statement, - ) -> Result, Error> { - match eql_mapper::type_check(self.context.get_table_resolver(), statement) { - Ok(typed_statement) => { - debug!(target: MAPPER, - client_id = self.context.client_id, - typed_statement = ?typed_statement - ); + resolver: &Arc, + ) -> Result, EqlMapperError> { + let typed_statement = eql_mapper::type_check(resolver.clone(), statement)?; + debug!(target: MAPPER, + client_id = self.context.client_id, + typed_statement = ?typed_statement + ); + Ok(typed_statement) + } - Ok(typed_statement) + /// The Proxy error for a statement that failed type checking. + /// `writes_encrypted` is whether it writes data while encrypted columns + /// are configured. + fn mapping_error(&self, err: EqlMapperError, writes_encrypted: bool) -> Error { + // A column this build cannot map is a refusal, not a coverage gap, + // so it gets its own error, which `Error::is_coverage_gap` rejects. + if let Some((table, column, column_type)) = err.as_unmappable_encrypted_column() { + warn!( + client_id = self.context.client_id, + msg = "Statement refused: it references a column declared with a legacy EQL v2 type that this build cannot encrypt or decrypt. Migrate the column to an EQL v3 domain type.", + table, + column, + column_type, + ); + counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); + return MappingError::UnmappableEncryptedColumn { + table: table.to_string(), + column: column.to_string(), + column_type: column_type.to_string(), } - // A column this build cannot map is a refusal, not a coverage gap, - // so it gets its own error: `must_fail_closed` keys off it to stop - // the passthrough fallback from serving the statement anyway. - Err(err) if err.as_unmappable_encrypted_column().is_some() => { - let (table, column, column_type) = err.as_unmappable_encrypted_column().unwrap(); - warn!( - client_id = self.context.client_id, - msg = "Statement refused: it references a column declared with a legacy EQL v2 type that this build cannot encrypt or decrypt. Migrate the column to an EQL v3 domain type.", - table, - column, - column_type, - ); - counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); - Err(MappingError::UnmappableEncryptedColumn { + .into(); + } + + // The type checker understood the statement and knows the encrypted + // column can't do it, so this is a refusal, not a coverage gap. + if err.is_unsatisfied_bounds() { + warn!( + client_id = self.context.client_id, + msg = "Statement refused: it uses an operation that an encrypted column's EQL type does not support.", + error = err.to_string(), + ); + counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); + return MappingError::UnsupportedEncryptedOperation(err.to_string()).into(); + } + + // The type checker never saw the table's columns, so a write to it + // could put plaintext in an encrypted column. + if let Some(unresolved) = err.as_unresolved_table().filter(|_| writes_encrypted) { + warn!( + client_id = self.context.client_id, + msg = "Statement refused: it writes data and refers to a table Proxy cannot resolve, so Proxy cannot tell whether it writes to an encrypted column.", + table = ?unresolved, + ); + counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); + return match unresolved { + UnresolvedTable::NotFound(table) => MappingError::UnresolvableTable { table: table.to_string(), - column: column.to_string(), - column_type: column_type.to_string(), - } - .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 { + }, + UnresolvedTable::Ambiguous(table) => MappingError::AmbiguousTable { table: table.to_string(), - } - .into()) + }, } - Err(EqlMapperError::InternalError(str)) => { + .into(); + } + + counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); + match err { + EqlMapperError::InternalError(str) => { warn!( client_id = self.context.client_id, msg = "Internal Error in EQL Mapper", mapping_errors_enabled = self.context.mapping_errors_enabled(), error = str, ); - counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); - Err(MappingError::Internal(str).into()) + MappingError::Internal(str).into() } - Err(err) => { + err => { warn!( client_id = self.context.client_id, msg = "Unmappable statement", mapping_errors_enabled = self.context.mapping_errors_enabled(), error = err.to_string(), ); - counter!(STATEMENTS_UNMAPPABLE_TOTAL).increment(1); - Err(MappingError::StatementCouldNotBeTypeChecked(err.to_string()).into()) + MappingError::StatementCouldNotBeTypeChecked(err.to_string()).into() } } } @@ -3704,7 +3755,9 @@ mod unresolvable_table_tests { } /// A connection whose schema has `encrypted (id, secret)` with `secret` - /// encrypted, and an `encrypted` encrypt config only when `encrypted` is set. + /// encrypted, and an `encrypted` encrypt config only when `encrypted` is set, + /// a native `notes (body)`, and when `encrypted` is set, `legacy (data)` with + /// `data` a legacy EQL v2 column. 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")); @@ -3722,6 +3775,19 @@ mod unresolvable_table_tests { eql_mapper::Column::native(secret) })); schema.add_table(table); + let mut notes = eql_mapper::Table::new(sqltk::parser::ast::Ident::new("notes")); + notes.add_column(Arc::new(eql_mapper::Column::native( + sqltk::parser::ast::Ident::new("body"), + ))); + schema.add_table(notes); + if encrypted { + let mut legacy = eql_mapper::Table::new(sqltk::parser::ast::Ident::new("legacy")); + legacy.add_column(Arc::new(eql_mapper::Column::unmappable_encrypted( + sqltk::parser::ast::Ident::new("data"), + "eql_v2_encrypted", + ))); + schema.add_table(legacy); + } let mut encryption = EncryptConfig::new(); if encrypted { encryption.insert( @@ -3856,6 +3922,7 @@ mod unresolvable_table_tests { let (mut frontend, _) = encrypted_client(); let sql = "CREATE TEMPORARY TABLE t (name text)"; assert_eq!(query(&mut frontend, sql).await, sql); + frontend.context.test_report_schema_execution_succeeded(); for sql in [ "INSERT INTO t (name) VALUES ('Ada')", "INSERT INTO pg_temp.t (name) VALUES ('Ada')", @@ -3868,53 +3935,110 @@ mod unresolvable_table_tests { #[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:?}"); + let sql = "CREATE TEMPORARY TABLE t (name text)"; + assert_eq!(query(&mut frontend, sql).await, sql); + frontend.context.test_report_schema_execution_succeeded(); + + let sql = "INSERT INTO t (name) VALUES ($1)"; + let result = frontend + .parse_handler( + operation_id(), + Parse { + statement: Bytes::from_static(b"insert"), + query: Bytes::copy_from_slice(sql.as_bytes()), + parameter_types: Vec::new(), + }, + ) + .await; + assert!(result.is_ok(), "{sql}: {result:?}"); + } + + /// PostgreSQL would write to the temporary table, but if Proxy's record of + /// it is stale the name means the permanent one. + #[tokio::test] + async fn write_to_a_name_shared_by_a_temporary_and_a_permanent_table_is_refused_as_ambiguous() { + let (mut frontend, _) = encrypted_client(); + let sql = "CREATE TEMPORARY TABLE notes (body text)"; + assert_eq!(query(&mut frontend, sql).await, sql); + frontend.context.test_report_schema_execution_succeeded(); + + let forwarded = query( + &mut frontend, + "INSERT INTO notes (body) VALUES ('plaintext')", + ) + .await; + + let ambiguous = MappingError::AmbiguousTable { + table: "notes".to_owned(), } + .to_string(); + assert!( + forwarded.contains(&ambiguous.replace('\'', "''''")) + && !forwarded.contains("plaintext"), + "{forwarded}" + ); + let sql = "INSERT INTO pg_temp.notes (body) VALUES ('Ada')"; + assert_eq!(query(&mut frontend, sql).await, sql); } - /// A table created by encryption-neutral DDL earlier in the same Query is - /// native, though the schema has not modelled it yet. + /// Only a coverage gap falls back, so a read that is refused for another + /// reason stays refused while mapping errors are off. #[tokio::test] - async fn write_to_a_table_created_earlier_in_the_query_is_forwarded() { + async fn read_of_a_legacy_column_is_refused_rather_than_falling_back() { 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); + assert!(!frontend.context.mapping_errors_enabled()); + + let forwarded = query(&mut frontend, "SELECT data FROM legacy").await; + + assert!( + forwarded.contains("RAISE EXCEPTION") && forwarded.contains("legacy EQL v2 type"), + "{forwarded}" + ); } - /// The exemption is for writes whose targets are all native tables, not - /// for any write that merely reads one. + /// The type checker understood the statement and knows the column can't do + /// it, so it is not a coverage gap and must not fall back with plaintext. #[tokio::test] - async fn native_temporary_table_does_not_excuse_an_encrypted_target() { + async fn read_using_an_operation_an_encrypted_column_does_not_support_is_refused() { let (mut frontend, _) = encrypted_client(); + assert!(!frontend.context.mapping_errors_enabled()); + let forwarded = query( &mut frontend, - "CREATE TEMPORARY TABLE t (name text); INSERT INTO encrypted (secret) SELECT 'plaintext' FROM t", + "SELECT id FROM encrypted WHERE secret > 'plaintext'", ) .await; + assert!( - forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + forwarded.contains("RAISE EXCEPTION") + && forwarded.contains("mapping-unsupported-encrypted-operation") + && !forwarded.contains("'plaintext'"), "{forwarded}" ); } + #[tokio::test] + async fn extended_read_using_an_operation_an_encrypted_column_does_not_support_is_refused() { + let (mut frontend, _) = encrypted_client(); + + let result = frontend + .parse_handler( + operation_id(), + parse("SELECT id FROM encrypted WHERE secret > $1"), + ) + .await; + + assert!( + matches!( + result, + Err(Error::Mapping(MappingError::UnsupportedEncryptedOperation( + _ + ))) + ), + "{result:?}" + ); + } + #[tokio::test] async fn native_temporary_table_does_not_excuse_another_table() { for sql in [ @@ -4022,6 +4146,8 @@ mod unresolvable_table_tests { assert_eq!(counter.encrypted.load(Ordering::SeqCst), 0); } + /// When every encrypted column lost its config to a domain conflict, the + /// encrypt config is empty, but the columns are still encrypted. #[tokio::test] async fn write_is_refused_when_only_columns_with_no_config_are_encrypted() { let mut schema = eql_mapper::Schema::new("public"); @@ -4030,6 +4156,9 @@ mod unresolvable_table_tests { sqltk::parser::ast::Ident::new(table_schema), sqltk::parser::ast::Ident::new("users"), ); + table.add_column(Arc::new(eql_mapper::Column::native( + sqltk::parser::ast::Ident::new("id"), + ))); table.add_column(Arc::new(eql_mapper::Column::eql( sqltk::parser::ast::Ident::new("email"), eql_mapper::EqlTraits::none(), @@ -4037,26 +4166,33 @@ mod unresolvable_table_tests { ))); 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, + let schema = Arc::new(schema); + + for sql in [ "INSERT INTO users (email) VALUES ('plaintext')", - ) - .await; - assert_refused(&forwarded, "users"); - assert_eq!(counter.encrypted.load(Ordering::SeqCst), 0); + "UPDATE app.users SET email = v.x FROM (VALUES ('plaintext')) AS v(x) WHERE app.users.id = 1", + "MERGE INTO app.users USING (SELECT 1 AS id) AS source ON app.users.id = source.id WHEN NOT MATCHED THEN INSERT (email) VALUES ('plaintext')", + "COPY (INSERT INTO app.users (email) VALUES ('plaintext') RETURNING id) TO STDOUT", + ] { + let (sender, _) = tokio::sync::mpsc::unbounded_channel(); + let context = Context::new( + 1, + Arc::new(TandemConfig::for_testing()), + Arc::new(EncryptConfig::new()), + schema.clone(), + Arc::new(rustls::RootCertStore::empty()), + Counter::default(), + sender, + ); + let mut frontend = Frontend::new(context); + + let forwarded = query(&mut frontend, sql).await; + + assert!( + forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{sql}: {forwarded}" + ); + } } #[tokio::test] @@ -4101,6 +4237,168 @@ mod unresolvable_table_tests { ); } + /// Writes EQL Mapper cannot type check, on a table it does resolve. + const UNCHECKED_WRITES: [&str; 4] = [ + "COPY (INSERT INTO encrypted (secret) VALUES ('plaintext') RETURNING id) TO STDOUT", + "MERGE INTO encrypted USING (SELECT 1 AS id) AS source ON encrypted.id = source.id WHEN NOT MATCHED THEN INSERT (secret) VALUES ('plaintext')", + "PREPARE p AS INSERT INTO encrypted (secret) VALUES ('plaintext')", + "PREPARE p (text) AS INSERT INTO encrypted (secret) VALUES ($1)", + ]; + + #[tokio::test] + async fn simple_write_that_fails_type_checking_is_refused() { + for sql in UNCHECKED_WRITES { + 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 extended_write_that_fails_type_checking_is_refused() { + for sql in UNCHECKED_WRITES { + let (mut frontend, _) = encrypted_client(); + let result = frontend.parse_handler(operation_id(), parse(sql)).await; + assert!( + matches!( + &result, + Err(Error::Mapping( + MappingError::StatementCouldNotBeTypeChecked(_) + )) + ), + "{sql}: {result:?}" + ); + } + } + + #[tokio::test] + async fn write_that_fails_type_checking_is_forwarded_without_encrypted_columns() { + for sql in UNCHECKED_WRITES { + let (mut frontend, _) = client(TandemConfig::for_testing(), false); + assert_eq!(query(&mut frontend, sql).await, sql); + } + } + + #[tokio::test] + async fn extended_read_on_an_unresolvable_table_is_forwarded() { + let (mut frontend, _) = encrypted_client(); + let sql = "SELECT * FROM pg_catalog.pg_type WHERE typname = $1"; + let result = frontend.parse_handler(operation_id(), parse(sql)).await; + assert!(matches!(result, Ok(None)), "{result:?}"); + } + + /// Every statement in a Query is routed, so one after a fallback is still + /// checked rather than forwarded with the rest of the Query. + #[tokio::test] + async fn statements_after_a_passthrough_fallback_are_checked() { + let (mut frontend, _) = encrypted_client(); + let forwarded = query( + &mut frontend, + "SELECT * FROM unknown_view; PREPARE p AS INSERT INTO encrypted (secret) VALUES ('plaintext')", + ) + .await; + assert!( + forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{forwarded}" + ); + } + + /// Each statement in a Query is type checked against the schema as the + /// encryption-neutral DDL before it leaves it. + #[tokio::test] + async fn write_to_a_native_table_changed_earlier_in_the_query_is_forwarded() { + for sql in [ + "CREATE TABLE names (name text); INSERT INTO names (name) VALUES ('Ada')", + "CREATE TEMPORARY TABLE names (name text NOT NULL UNIQUE); INSERT INTO names (name) VALUES ('Ada'), ('Grace')", + "CREATE TABLE public.names (name text); INSERT INTO public.names (name) VALUES ('Ada')", + "CREATE TABLE names (name text); UPDATE names SET name = 'Ada'; DELETE FROM names", + "ALTER TABLE encrypted ADD COLUMN note text; INSERT INTO encrypted (id, note) VALUES (1, 'Ada')", + "CREATE TABLE names (name text); ALTER TABLE names RENAME TO people; INSERT INTO people (name) VALUES ('Ada')", + "CREATE TEMPORARY TABLE names (name text); ALTER TABLE names ADD COLUMN nickname text; INSERT INTO names (nickname) VALUES ('Ada')", + "CREATE TABLE names (name text); CREATE TABLE IF NOT EXISTS names (other int); INSERT INTO names (name) VALUES ('Ada')", + "CREATE TABLE IF NOT EXISTS notes (other int); INSERT INTO notes (body) VALUES ('Ada')", + "CREATE TEMPORARY TABLE IF NOT EXISTS names (name text); INSERT INTO names (name) VALUES ('Ada')", + // PostgreSQL creates a table in `pg_temp` as temporary, without the keyword. + "CREATE TABLE pg_temp.names (name text); INSERT INTO names (name) VALUES ('Ada')", + ] { + let (mut frontend, _) = encrypted_client(); + assert_eq!(query(&mut frontend, sql).await, sql, "{sql}"); + } + } + + /// `IF NOT EXISTS` leaves an existing table as it is, so a batch view can't + /// assume the columns it declares when the table may hold encrypted ones. + #[tokio::test] + async fn write_after_create_if_not_exists_of_an_encrypted_table_is_not_forwarded_as_plaintext() + { + let (mut frontend, _) = encrypted_client(); + let forwarded = query( + &mut frontend, + "CREATE TABLE IF NOT EXISTS encrypted (id int, secret text); INSERT INTO encrypted (secret) VALUES ('plaintext')", + ) + .await; + + assert!(!forwarded.contains("'plaintext'"), "{forwarded}"); + } + + #[tokio::test] + async fn write_that_is_not_only_to_a_table_created_earlier_in_the_query_is_refused() { + for sql in [ + // Created after the write, not before it. + "INSERT INTO names (name) VALUES ('plaintext'); CREATE TABLE names (name text)", + // A different table from the one created. + "CREATE TABLE names (name text); INSERT INTO accounts (secret) VALUES ('plaintext')", + // Created after the write, not before it. + "INSERT INTO names (name) VALUES ('plaintext'); CREATE TEMPORARY TABLE names (name text)", + // `IF NOT EXISTS` skips the create when the table exists, and Proxy + // may not have loaded it, so its columns may be encrypted. + "CREATE TABLE IF NOT EXISTS late (secret text); INSERT INTO late (secret) VALUES ('plaintext')", + "CREATE TABLE IF NOT EXISTS public.late (secret text); INSERT INTO late (secret) VALUES ('plaintext')", + // Created after the write, with other DDL before the write. + "CREATE TABLE other (id int); INSERT INTO names (name) VALUES ('plaintext'); CREATE TEMPORARY TABLE names (name text)", + // Dropped before the write. + "CREATE TEMPORARY TABLE names (name text); DROP TABLE names; INSERT INTO names (name) VALUES ('plaintext')", + // A nested write to a table Proxy can't resolve. + "CREATE TABLE names (name text); WITH w AS (INSERT INTO accounts (secret) VALUES ('plaintext') RETURNING 1) INSERT INTO names SELECT 'x' FROM w", + // Qualified differently from the table created. + "CREATE TABLE archive.names (name text); INSERT INTO names (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 write_to_an_encrypted_table_reading_a_table_created_earlier_in_the_query_is_encrypted() + { + let (mut frontend, _) = encrypted_client(); + let forwarded = query( + &mut frontend, + "CREATE TABLE names (name text); INSERT INTO encrypted (secret) SELECT 'plaintext' FROM names", + ) + .await; + + assert!( + !forwarded.contains("RAISE EXCEPTION") && !forwarded.contains("plaintext"), + "{forwarded}" + ); + } + + fn parse(sql: &str) -> Parse { + Parse { + statement: Bytes::from_static(b"s"), + query: Bytes::copy_from_slice(sql.as_bytes()), + parameter_types: Vec::new(), + } + } + #[tokio::test] async fn passthrough_fallback_still_serves_a_query_with_nothing_encrypted() { let (mut frontend, _) = encrypted_client(); diff --git a/packages/cipherstash-proxy/src/proxy/schema/middleware.rs b/packages/cipherstash-proxy/src/proxy/schema/middleware.rs index e000c476a..758f26230 100644 --- a/packages/cipherstash-proxy/src/proxy/schema/middleware.rs +++ b/packages/cipherstash-proxy/src/proxy/schema/middleware.rs @@ -43,7 +43,8 @@ //! encryption metadata and then perform schema-dependent work are rejected. Explicitly native //! table DDL remains compatible because it cannot create an encryption obligation. //! * Temporary and other connection-local catalog objects are invisible to the separate -//! publication connection. Native temporary-table batches may pass through, but encrypted +//! publication connection. Native temporary tables are modelled per connection instead: +//! they survive commit, follow rollback and savepoints, and are never published. Encrypted //! temporary objects cannot be represented or authoritatively published and are unsupported. //! A temporary table that could shadow encrypted metadata leaves the connection fail-closed //! until an earlier savepoint removes it or the connection ends; an authoritative global reload @@ -69,10 +70,10 @@ use crate::proxy::EncryptConfig; use cipherstash_client::eql::Identifier; use eql_mapper::{ColumnKind, Schema, SchemaWithEdits, TableResolver}; use sqltk::parser::ast::{ - AlterColumnOperation, AlterTableOperation, ColumnDef, DataType, DropBehavior, Ident, - ObjectName, ObjectNamePart, ObjectType, Statement, + AlterColumnOperation, AlterTableOperation, ColumnDef, DataType, DiscardObject, DropBehavior, + Ident, ObjectName, ObjectNamePart, ObjectType, OnCommit, Statement, }; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, VecDeque}; use std::sync::{ atomic::{AtomicBool, AtomicUsize, Ordering}, Arc, RwLock, @@ -85,6 +86,172 @@ struct Intent { statement: Statement, ddl: bool, modelled: bool, + /// Changes only this connection's native temporary tables. + temporary: bool, +} + +impl Intent { + /// Whether schema-dependent work must wait for this execution's outcome: + /// catalog DDL, or a temporary table a later statement may use. Work after + /// `DISCARD` need not wait, since a temporary table it removes can only + /// make a later statement resolve a table that PostgreSQL then rejects. + fn changes_schema(&self) -> bool { + self.ddl || (self.temporary && matches!(self.statement, Statement::CreateTable(_))) + } +} + +/// Where one confirmed schema change lands. +enum SchemaChange { + /// DDL on this connection's native temporary tables. + Temporary(Statement), + /// `DISCARD TEMP` or `DISCARD ALL`. + DiscardTemporary, + /// DDL on catalog tables. + Catalog(Statement), + /// A change Proxy cannot model. + Unmodelled { connection_local: bool }, +} + +/// A connection's native temporary tables, which outlive its transactions +/// but not a rollback of the transaction that created them. +#[derive(Debug)] +struct Temporary { + /// The tables now, for DDL to edit and the type checker to read. + resolver: Arc, + /// Tables created `ON COMMIT DROP` in the current transaction. + on_commit_drop: Vec, + /// The tables when the current transaction started, which rollback restores. + committed: SchemaWithEdits, + /// Whether PostgreSQL will roll back the current transaction. + failed: bool, +} + +#[derive(Clone, Debug)] +/// Temporary tables as of a savepoint. +struct TemporaryCheckpoint { + tables: SchemaWithEdits, + on_commit_drop: Vec, +} + +impl Temporary { + fn new() -> Self { + let resolver = Arc::new(TableResolver::new_temporary()); + let committed = Self::tables_of(&resolver); + Self { + resolver, + on_commit_drop: Vec::new(), + committed, + failed: false, + } + } + + fn tables_of(resolver: &TableResolver) -> SchemaWithEdits { + resolver + .as_schema_with_edits() + .unwrap() + .read() + .unwrap() + .clone() + } + + /// Whether `name` resolves to one of these tables. + fn checkpoint(&self) -> TemporaryCheckpoint { + TemporaryCheckpoint { + tables: Self::tables_of(&self.resolver), + on_commit_drop: self.on_commit_drop.clone(), + } + } + + fn restore(&mut self, checkpoint: TemporaryCheckpoint) { + *self + .resolver + .as_schema_with_edits() + .unwrap() + .write() + .unwrap() = checkpoint.tables; + self.on_commit_drop = checkpoint.on_commit_drop; + } + + /// Applies a confirmed `CREATE TEMPORARY TABLE`, `ALTER TABLE` or + /// `DROP TABLE` naming these tables. + fn apply(&mut self, statement: &Statement) { + apply_to_temporary(&self.resolver, statement); + if let Statement::CreateTable(create) = statement { + if create.on_commit == Some(OnCommit::Drop) { + self.on_commit_drop.push(create.name.clone()); + } + } + } + + fn discard(&mut self) { + self.restore(TemporaryCheckpoint { + tables: Self::tables_of(&TableResolver::new_temporary()), + on_commit_drop: Vec::new(), + }); + } + + /// Commits the current transaction's changes, or discards them if + /// PostgreSQL rolled it back. + fn transaction_ended(&mut self) { + if self.failed { + self.restore(TemporaryCheckpoint { + tables: self.committed.clone(), + on_commit_drop: Vec::new(), + }); + } else { + for name in std::mem::take(&mut self.on_commit_drop) { + apply_to_temporary(&self.resolver, &drop_tables(vec![name])); + } + } + self.committed = Self::tables_of(&self.resolver); + self.failed = false; + } +} + +/// Applies DDL to a temporary-table resolver. A `DROP` removes the tables +/// whatever its drop behaviour, since nothing Proxy models depends on them. +fn apply_to_temporary(resolver: &Arc, statement: &Statement) { + let statement = match statement { + Statement::Drop { + object_type: ObjectType::Table, + names, + .. + } => drop_tables(names.clone()), + statement => statement.clone(), + }; + eql_mapper::collect_ddl_with_column_kind(resolver.clone(), &statement, &column_kind); +} + +/// An editable copy of `resolver`'s tables that later edits to either leave +/// the other unchanged. +fn snapshot(resolver: &TableResolver) -> Arc { + Arc::new(TableResolver::ViaSchemaWithEdits(Arc::new(RwLock::new( + Temporary::tables_of(resolver), + )))) +} + +/// `statement`, with a `CREATE TABLE` in `pg_temp` marked temporary, since +/// PostgreSQL creates any table there as a temporary table. +fn mark_temporary(mut statement: Statement) -> Statement { + if let Statement::CreateTable(create) = &mut statement { + if let [ObjectNamePart::Identifier(qualifier), _] = create.name.0.as_slice() { + create.temporary |= postgres_identifier(qualifier) == "pg_temp"; + } + } + statement +} + +/// `DROP TABLE` of `names`. +fn drop_tables(names: Vec) -> Statement { + Statement::Drop { + object_type: ObjectType::Table, + if_exists: false, + names, + cascade: false, + restrict: false, + purge: false, + temporary: false, + } } #[derive(Clone, Debug)] @@ -104,6 +271,7 @@ struct Savepoint { encrypt_config: EncryptConfig, unmodelled: bool, local_unmodelled: bool, + temporary: TemporaryCheckpoint, } /// PostgreSQL's transaction state carried by a `ReadyForQuery` message. @@ -158,12 +326,7 @@ 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>>, + temporary: Arc>, } impl SchemaMiddleware { @@ -196,7 +359,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())), + temporary: Arc::new(RwLock::new(Temporary::new())), } } @@ -205,6 +368,15 @@ impl SchemaMiddleware { self.resolver.read().unwrap().clone() } + /// Returns the resolver the type checker uses: this connection's native + /// temporary tables in front of its effective schema. + pub fn effective_resolver(&self) -> Arc { + Arc::new(TableResolver::with_temporary( + self.temporary.read().unwrap().resolver.clone(), + self.resolver(), + )) + } + /// Whether the loaded schema has any encrypted column, whether or not the /// encrypt config names it. pub fn schema_has_encrypted_columns(&self) -> bool { @@ -278,60 +450,41 @@ impl SchemaMiddleware { self.discard_skipped_executions(); self.transaction_active .store(!status.is_idle(), Ordering::Release); + if status.is_idle() { + self.temporary.write().unwrap().transaction_ended(); + } if status.is_idle() && !self.needs_publication() { self.adopt_latest(); } } - /// 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)); - } - } + /// Classifies what executing `statement` would do to schema state. + fn intent(&self, statement: Statement) -> Intent { + let statement = mark_temporary(statement); + Intent { + ddl: self.is_schema_ddl(&statement), + modelled: self.models_ddl(&statement), + temporary: self.changes_only_temporary_tables(&statement), + statement, } } - /// 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", + /// Whether `statement` changes only native temporary tables: creates one + /// that cannot shadow an encrypted table, or discards them all. + fn changes_only_temporary_tables(&self, statement: &Statement) -> bool { + match statement { + Statement::CreateTable(create) if create.temporary => !self.is_schema_ddl(statement), + Statement::Discard { + object_type: DiscardObject::TEMP | DiscardObject::ALL, + } => true, _ => 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 { - ddl: self.is_schema_ddl(&statement), - modelled: self.models_ddl(&statement), - statement, - }, - ); + let intent = self.intent(statement); + self.prepared.write().unwrap().insert(name, intent); } /// Associates a portal with the schema intent of its prepared statement. @@ -357,7 +510,7 @@ impl SchemaMiddleware { /// the protocol adapter must flush DDL immediately to unblock dependent work. pub fn execute(&self, portal: &Name) -> bool { let intent = self.portals.read().unwrap().get(portal).cloned(); - let executes_ddl = intent.as_ref().is_some_and(|intent| intent.ddl); + let executes_ddl = intent.as_ref().is_some_and(Intent::changes_schema); self.transaction_active.store(true, Ordering::Release); if executes_ddl { self.in_flight_ddl.fetch_add(1, Ordering::AcqRel); @@ -371,16 +524,11 @@ 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 { - let intent = Intent { - ddl: self.is_schema_ddl(statement), - modelled: self.models_ddl(statement), - statement: statement.clone(), - }; - if intent.ddl { + let intent = self.intent(statement.clone()); + if intent.changes_schema() { self.in_flight_ddl.fetch_add(1, Ordering::AcqRel); } executions.push_back(PendingExecution::Execute(Some(Box::new(intent)))); @@ -542,25 +690,21 @@ impl SchemaMiddleware { #[cfg(test)] /// Records a direct statement execution for state-machine tests. pub fn execution_started(&self, statement: Statement) { - let ddl = self.is_schema_ddl(&statement); - if ddl { + let intent = self.intent(statement); + if intent.changes_schema() { self.in_flight_ddl.fetch_add(1, Ordering::AcqRel); } self.executions .write() .unwrap() - .push_back(PendingExecution::Execute(Some(Box::new(Intent { - modelled: self.models_ddl(&statement), - statement, - ddl, - })))); + .push_back(PendingExecution::Execute(Some(Box::new(intent)))); } /// Applies the next queued execution after PostgreSQL confirms success. pub fn execution_succeeded(&self) { if let Some(Some(intent)) = self.pop_execution() { - let statement = intent.statement; - match statement { + let changes_schema = intent.changes_schema(); + match intent.statement.clone() { Statement::Savepoint { name } => { let resolver = self.resolver(); let overlay = resolver.as_schema_with_edits().unwrap(); @@ -571,6 +715,7 @@ impl SchemaMiddleware { encrypt_config: (*self.encrypt_config()).clone(), unmodelled: self.unmodelled.load(Ordering::Acquire), local_unmodelled: self.local_unmodelled.load(Ordering::Acquire), + temporary: self.temporary.read().unwrap().checkpoint(), }); } Statement::ReleaseSavepoint { name } => { @@ -597,6 +742,10 @@ impl SchemaMiddleware { let encrypt_config = savepoints[index].encrypt_config.clone(); let unmodelled = savepoints[index].unmodelled; let local_unmodelled = savepoints[index].local_unmodelled; + let mut temporary = self.temporary.write().unwrap(); + temporary.restore(savepoints[index].temporary.clone()); + temporary.failed = false; + drop(temporary); savepoints.truncate(index + 1); let resolver = self.resolver(); let overlay = resolver.as_schema_with_edits().unwrap(); @@ -624,21 +773,37 @@ impl SchemaMiddleware { self.savepoints.write().unwrap().clear(); self.dirty.store(false, Ordering::Release); self.unmodelled.store(false, Ordering::Release); + self.temporary.write().unwrap().failed = true; } - statement if intent.ddl => { - if !intent.modelled { - let local_only = matches!( - statement, - Statement::CreateTable(ref create) if create.temporary - ); - self.mark_unmodelled(local_only); - } else { - self.apply_ddl(&statement); + _ if intent.ddl || intent.temporary => { + let temporary = self.temporary.read().unwrap().resolver.clone(); + for change in self.split_change(&temporary, &intent.statement) { + match change { + SchemaChange::Temporary(statement) => { + self.temporary.write().unwrap().apply(&statement) + } + SchemaChange::DiscardTemporary => { + self.temporary.write().unwrap().discard() + } + SchemaChange::Catalog(statement) if self.models_ddl(&statement) => { + self.apply_ddl(&statement) + } + SchemaChange::Catalog(statement) => { + let local_only = matches!( + statement, + Statement::CreateTable(ref create) if create.temporary + ); + self.mark_unmodelled(local_only); + } + SchemaChange::Unmodelled { connection_local } => { + self.mark_unmodelled(connection_local) + } + } } } _ => {} } - if intent.ddl { + if changes_schema { self.dirty.store( self.unmodelled.load(Ordering::Acquire) || self.resolver().has_schema_changed(), Ordering::Release, @@ -649,6 +814,116 @@ impl SchemaMiddleware { } } + /// Splits the change `statement` makes between this connection's + /// temporary tables, which `temporary` resolves, and the catalog. A + /// `DROP TABLE` or `ALTER TABLE` naming a temporary table changes the + /// temporary table, as it does in PostgreSQL. + fn split_change(&self, temporary: &TableResolver, statement: &Statement) -> Vec { + if self.changes_only_temporary_tables(statement) { + return vec![match statement { + Statement::Discard { .. } => SchemaChange::DiscardTemporary, + statement => SchemaChange::Temporary(statement.clone()), + }]; + } + if !self.is_schema_ddl(statement) { + return vec![]; + } + let is_temporary = |name: &ObjectName| temporary.resolve_table(name).is_ok(); + match statement { + Statement::Drop { + object_type: ObjectType::Table, + names, + .. + } => { + let (temporary_names, catalog_names): (Vec<_>, Vec<_>) = + names.iter().cloned().partition(is_temporary); + let mut changes = vec![]; + if !temporary_names.is_empty() { + changes.push(SchemaChange::Temporary(drop_tables(temporary_names))); + } + if !catalog_names.is_empty() { + let mut statement = statement.clone(); + if let Statement::Drop { names, .. } = &mut statement { + *names = catalog_names; + } + changes.push(SchemaChange::Catalog(statement)); + } + changes + } + Statement::AlterTable { name, .. } if is_temporary(name) => { + vec![if self.ddl_may_change_encryption(statement) { + SchemaChange::Unmodelled { + connection_local: true, + } + } else { + SchemaChange::Temporary(statement.clone()) + }] + } + statement => vec![SchemaChange::Catalog(statement.clone())], + } + } + + /// The resolver to type check each statement of a simple-query batch with: + /// the effective schema plus the encryption-neutral schema changes of the + /// statements before it. PostgreSQL stops a batch at its first error, so a + /// statement runs only once the changes before it have been made. A batch + /// that changes encryption metadata before a dependent statement is refused + /// by [`Self::simple_query_requires_fail_closed`]. + pub fn batch_resolvers(&self, statements: &[Statement]) -> Vec> { + let mut resolver = self.effective_resolver(); + let mut scratch: Option<(Arc, Arc)> = None; + statements + .iter() + .map(|statement| { + let current = resolver.clone(); + let statement = &mark_temporary(statement.clone()); + if self.ddl_may_change_encryption(statement) { + return current; + } + let changes = match &scratch { + Some((temporary, _)) => self.split_change(temporary, statement), + None => self.split_change(&self.temporary.read().unwrap().resolver, statement), + }; + if changes.is_empty() { + return current; + } + let (temporary, catalog) = scratch.get_or_insert_with(|| self.scratch()); + for change in changes { + match change { + SchemaChange::Temporary(statement) => { + apply_to_temporary(temporary, &statement) + } + SchemaChange::DiscardTemporary => { + *temporary = Arc::new(TableResolver::new_temporary()) + } + SchemaChange::Catalog(statement) if self.models_ddl(&statement) => { + eql_mapper::collect_ddl_with_column_kind( + catalog.clone(), + &statement, + &column_kind, + ); + } + SchemaChange::Catalog(_) | SchemaChange::Unmodelled { .. } => {} + } + } + resolver = Arc::new(TableResolver::with_temporary( + snapshot(temporary), + snapshot(catalog), + )); + current + }) + .collect() + } + + /// Copies of the temporary and catalog layers for a batch to change + /// without changing this connection's state. + fn scratch(&self) -> (Arc, Arc) { + ( + snapshot(&self.temporary.read().unwrap().resolver), + snapshot(&self.resolver()), + ) + } + /// Applies modelled DDL to schema and encryption overlays as one operation. fn apply_ddl(&self, statement: &Statement) { eql_mapper::collect_ddl_with_column_kind(self.resolver(), statement, &|column| { @@ -662,8 +937,9 @@ impl SchemaMiddleware { /// Discards the next queued execution after PostgreSQL reports failure. pub fn execution_failed(&self) { + self.temporary.write().unwrap().failed = true; if let Some(Some(intent)) = self.pop_execution() { - if intent.ddl { + if intent.changes_schema() { self.in_flight_ddl.fetch_sub(1, Ordering::AcqRel); self.execution_finished.notify_waiters(); } @@ -689,7 +965,9 @@ impl SchemaMiddleware { let mut executions = self.executions.write().unwrap(); while let Some(execution) = executions.pop_front() { match execution { - PendingExecution::Execute(Some(intent)) if intent.ddl => discarded_ddl += 1, + PendingExecution::Execute(Some(intent)) if intent.changes_schema() => { + discarded_ddl += 1 + } PendingExecution::Execute(_) => {} PendingExecution::ReadyBoundary => break, } @@ -1655,4 +1933,268 @@ mod tests { .resolve_table(&table("later_table")) .is_ok()); } + + /// Executes `sql` and reports that PostgreSQL succeeded. + fn run(middleware: &SchemaMiddleware, sql: &str) { + middleware.execution_started(parse(sql)); + middleware.execution_succeeded(); + } + + fn resolves(middleware: &SchemaMiddleware, name: &str) -> bool { + middleware + .effective_resolver() + .resolve_table(&table(name)) + .is_ok() + } + + fn pg_temp(name: &str) -> ObjectName { + ObjectName(vec![ + ObjectNamePart::Identifier(Ident::new("pg_temp")), + ObjectNamePart::Identifier(Ident::new(name)), + ]) + } + + #[test] + fn a_temporary_table_resolves_once_its_create_succeeds() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + + middleware.execution_started(parse("create temporary table names (name text)")); + assert!(!resolves(&middleware, "names")); + + middleware.execution_succeeded(); + assert!(resolves(&middleware, "names")); + assert!(middleware + .effective_resolver() + .resolve_table(&pg_temp("names")) + .is_ok()); + assert!(!middleware.has_unmodelled_ddl()); + assert!(!middleware.needs_publication()); + } + + #[test] + fn a_table_created_in_pg_temp_is_temporary() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + + run(&middleware, "create table pg_temp.names (name text)"); + + assert!(resolves(&middleware, "names")); + assert!(!middleware.has_unmodelled_ddl()); + assert!(!middleware.needs_publication()); + } + + #[test] + fn a_temporary_table_whose_create_failed_does_not_resolve() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + + middleware.execution_started(parse("create temporary table names (name text)")); + middleware.execution_failed(); + + assert!(!resolves(&middleware, "names")); + } + + #[tokio::test] + async fn dependent_mapping_waits_for_a_temporary_table_create() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + middleware.execution_started(parse("create temporary table names (name text)")); + + assert!(tokio::time::timeout( + std::time::Duration::from_millis(20), + middleware.wait_for_ddl(), + ) + .await + .is_err()); + + middleware.execution_succeeded(); + middleware.wait_for_ddl().await; + assert!(resolves(&middleware, "names")); + } + + #[test] + fn a_temporary_table_survives_commit_and_later_transactions() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + + run(&middleware, "create temporary table names (name text)"); + middleware.ready_for_query(TransactionStatus::Idle); + middleware.publication_succeeded(); + run(&middleware, "begin"); + run(&middleware, "commit"); + middleware.ready_for_query(TransactionStatus::Idle); + + assert!(resolves(&middleware, "names")); + } + + #[test] + fn dropping_a_temporary_table_removes_it_without_touching_the_catalog() { + for drop in ["drop table names", "drop table pg_temp.names"] { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "create temporary table names (name text)"); + + run(&middleware, drop); + + assert!(!resolves(&middleware, "names"), "{drop}"); + assert!(!middleware.has_unmodelled_ddl(), "{drop}"); + assert!(!middleware.needs_publication(), "{drop}"); + } + } + + #[test] + fn altering_a_temporary_table_changes_it_without_touching_the_catalog() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "create temporary table names (name text)"); + + run(&middleware, "alter table names add column nickname text"); + + assert!(middleware + .effective_resolver() + .resolve_table_column(&table("names"), &Ident::new("nickname")) + .is_ok()); + assert!(!middleware.has_unmodelled_ddl()); + assert!(!middleware.needs_publication()); + } + + #[test] + fn discard_removes_temporary_tables() { + for discard in ["discard temp", "discard all"] { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "create temporary table names (name text)"); + + run(&middleware, discard); + + assert!(!resolves(&middleware, "names"), "{discard}"); + } + } + + #[test] + fn rollback_removes_a_temporary_table_created_in_the_transaction() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "create temporary table kept (name text)"); + middleware.ready_for_query(TransactionStatus::Idle); + + run(&middleware, "begin"); + run(&middleware, "create temporary table names (name text)"); + run(&middleware, "drop table kept"); + run(&middleware, "rollback"); + middleware.ready_for_query(TransactionStatus::Idle); + + assert!(!resolves(&middleware, "names")); + assert!(resolves(&middleware, "kept")); + } + + #[test] + fn rollback_to_savepoint_restores_temporary_tables() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "begin"); + run(&middleware, "create temporary table kept (name text)"); + run(&middleware, "savepoint before_names"); + run(&middleware, "create temporary table names (name text)"); + run(&middleware, "rollback to savepoint before_names"); + run(&middleware, "commit"); + middleware.ready_for_query(TransactionStatus::Idle); + + assert!(!resolves(&middleware, "names")); + assert!(resolves(&middleware, "kept")); + } + + /// A batch runs in one implicit transaction, so an error later in it + /// undoes a temporary table an earlier statement created. + #[test] + fn a_failed_transaction_removes_its_temporary_tables() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "create temporary table names (name text)"); + middleware.execution_started(parse("select 1 / 0")); + middleware.execution_failed(); + middleware.ready_for_query(TransactionStatus::Idle); + + assert!(!resolves(&middleware, "names")); + } + + #[test] + fn a_failure_undone_by_rollback_to_savepoint_keeps_temporary_tables() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "begin"); + run(&middleware, "create temporary table names (name text)"); + run(&middleware, "savepoint before_error"); + middleware.execution_started(parse("select 1 / 0")); + middleware.execution_failed(); + middleware.ready_for_query(TransactionStatus::FailedTransaction); + run(&middleware, "rollback to savepoint before_error"); + run(&middleware, "commit"); + middleware.ready_for_query(TransactionStatus::Idle); + + assert!(resolves(&middleware, "names")); + } + + #[test] + fn on_commit_drop_removes_a_temporary_table_at_commit() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + run(&middleware, "begin"); + run( + &middleware, + "create temporary table names (name text) on commit drop", + ); + middleware.ready_for_query(TransactionStatus::InTransaction); + assert!(resolves(&middleware, "names")); + + run(&middleware, "commit"); + middleware.ready_for_query(TransactionStatus::Idle); + + assert!(!resolves(&middleware, "names")); + } + + /// DDL that may change encryption is left out of a batch view, so a + /// statement after it does not type check against encryption Proxy has + /// not loaded. + #[test] + fn a_batch_view_leaves_out_ddl_that_may_change_encryption() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + let statements = Parser::parse_sql( + &PostgreSqlDialect {}, + "create table secrets (secret eql_v3_text); + insert into secrets (secret) values ('a')", + ) + .unwrap(); + + let views = middleware.batch_resolvers(&statements); + + assert!(views[1].resolve_table(&table("secrets")).is_err()); + } + + /// Each statement in a batch sees the DDL before it, and none after it. + #[test] + fn a_batch_view_reflects_only_earlier_statements() { + let middleware = SchemaMiddleware::new(Arc::new(Schema::new("public"))); + let statements = Parser::parse_sql( + &PostgreSqlDialect {}, + "create table early (id int); + insert into names (name) values ('a'); + create temporary table names (name text); + drop table early; + insert into names (name) values ('b')", + ) + .unwrap(); + + let views = middleware + .batch_resolvers(&statements) + .iter() + .map(|resolver| { + ( + resolver.resolve_table(&table("early")).is_ok(), + resolver.resolve_table(&table("names")).is_ok(), + ) + }) + .collect::>(); + + assert_eq!( + views, + vec![ + (false, false), + (true, false), + (true, false), + (true, true), + (false, true), + ] + ); + assert!(!resolves(&middleware, "early")); + assert!(!resolves(&middleware, "names")); + } } diff --git a/packages/eql-mapper/src/eql_mapper.rs b/packages/eql-mapper/src/eql_mapper.rs index 435fb8d7b..b3e2dddd5 100644 --- a/packages/eql-mapper/src/eql_mapper.rs +++ b/packages/eql-mapper/src/eql_mapper.rs @@ -3,7 +3,7 @@ use crate::{ inference::{TypeError, TypeInferencer}, unifier::{EqlTerm, Projection, Type, Unifier, Value}, DepMut, Param, ParamError, ScopeError, ScopeTracker, TableResolver, TypeCheckedStatement, - TypeRegistry, + TypeRegistry, UnresolvedTable, }; use sqltk::parser::ast::{self as ast, Statement}; use sqltk::{Break, NodeKey, Visitable, Visitor}; @@ -95,57 +95,6 @@ pub fn writes_data(statement: &Statement) -> bool { 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 { @@ -196,19 +145,32 @@ impl EqlMapperError { } } - /// Returns the table name when this error means the statement names a + /// Whether the statement uses an operation that an encrypted value's EQL + /// type does not support. The type checker understood the statement, so + /// this is not a gap in its coverage. + pub fn is_unsatisfied_bounds(&self) -> bool { + match self { + EqlMapperError::Type(err) => err.is_unsatisfied_bounds(), + EqlMapperError::Scope(err) => err.is_unsatisfied_bounds(), + EqlMapperError::Import(ImportError::ScopeError(err)) => err.is_unsatisfied_bounds(), + EqlMapperError::Import(ImportError::TypeError(err)) => err.is_unsatisfied_bounds(), + _ => false, + } + } + + /// Returns why, 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> { + pub fn as_unresolved_table(&self) -> Option> { 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(), + EqlMapperError::Type(err) => err.as_unresolved_table(), + EqlMapperError::Scope(err) => err.as_unresolved_table(), + EqlMapperError::Import(ImportError::SchemaError(err)) => err.as_unresolved_table(), + EqlMapperError::Import(ImportError::ScopeError(err)) => err.as_unresolved_table(), + EqlMapperError::Import(ImportError::TypeError(err)) => err.as_unresolved_table(), _ => None, } } diff --git a/packages/eql-mapper/src/inference/type_error.rs b/packages/eql-mapper/src/inference/type_error.rs index fdbd47be0..ad0aefb92 100644 --- a/packages/eql-mapper/src/inference/type_error.rs +++ b/packages/eql-mapper/src/inference/type_error.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use crate::{unifier::Type, SchemaError, ScopeError}; +use crate::{unifier::Type, SchemaError, ScopeError, UnresolvedTable}; use super::unifier::EqlTraits; @@ -140,12 +140,23 @@ impl TypeError { } } - /// 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> { + /// Whether this error means a statement uses an operation that an encrypted + /// value's EQL type does not support, such as `<` on a column with no + /// ordering index. + pub fn is_unsatisfied_bounds(&self) -> bool { match self { - TypeError::SchemaError(err) => err.as_table_not_found(), - TypeError::ScopeError(err) => err.as_table_not_found(), + TypeError::UnsatisfiedBounds(..) => true, + TypeError::ScopeError(err) => err.is_unsatisfied_bounds(), + _ => false, + } + } + + /// Returns why, when this error means a statement names a + /// table the schema cannot resolve. + pub fn as_unresolved_table(&self) -> Option> { + match self { + TypeError::SchemaError(err) => err.as_unresolved_table(), + TypeError::ScopeError(err) => err.as_unresolved_table(), _ => None, } } diff --git a/packages/eql-mapper/src/lib.rs b/packages/eql-mapper/src/lib.rs index 1443a3bb5..c4a977e9e 100644 --- a/packages/eql-mapper/src/lib.rs +++ b/packages/eql-mapper/src/lib.rs @@ -41,7 +41,9 @@ pub(crate) use transformation_rules::*; #[cfg(test)] mod test { - use super::{test_helpers::*, type_check, EqlMapperError, ScopeError, TypeError}; + use super::{ + test_helpers::*, type_check, EqlMapperError, ScopeError, TypeError, UnresolvedTable, + }; use crate::{ projection, schema, test_helpers, unifier::{ @@ -540,10 +542,10 @@ mod test { 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(), + Err(err) => match err.as_unresolved_table() { + Some(UnresolvedTable::NotFound(table)) => table.to_owned(), + _ => panic!("failed for another reason ({err}): {sql}"), + }, } } @@ -678,37 +680,6 @@ mod test { } } - #[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 390eef956..4f9831361 100644 --- a/packages/eql-mapper/src/model/schema.rs +++ b/packages/eql-mapper/src/model/schema.rs @@ -109,20 +109,37 @@ pub enum SchemaError { #[error("Table not found: {}", _0)] TableNotFound(String), - #[error("Column: {} not found for table: {}", _0, _1)] + #[error( + "Table name is ambiguous: {} names both a temporary and a permanent table", + _0 + )] + AmbiguousTable(String), + + #[error("Column: {} not found for table: {}", _1, _0)] 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> { + /// Returns why a table could not be resolved, when that is what this + /// error means. + pub fn as_unresolved_table(&self) -> Option> { match self { - SchemaError::TableNotFound(table) => Some(table), + SchemaError::TableNotFound(table) => Some(UnresolvedTable::NotFound(table)), + SchemaError::AmbiguousTable(table) => Some(UnresolvedTable::Ambiguous(table)), SchemaError::ColumnNotFound(..) => None, } } } +/// A table name that does not resolve to exactly one table. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UnresolvedTable<'a> { + /// No table has this name. + NotFound(&'a str), + /// A temporary table and a permanent table both have this bare name. + Ambiguous(&'a str), +} + impl Schema { /// Whether any column holds encrypted data, including one this build /// cannot map. @@ -449,3 +466,23 @@ macro_rules! schema { } }; } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn column_not_found_names_the_column_then_the_table() { + let mut schema = Schema::new("public"); + schema.add_table(Table::new(Ident::new("users"))); + + let err = schema + .resolve_table_column( + &ObjectName::from(vec![Ident::new("users")]), + &Ident::new("email"), + ) + .unwrap_err(); + + assert_eq!(err.to_string(), "Column: email not found for table: users"); + } +} diff --git a/packages/eql-mapper/src/model/table_resolver.rs b/packages/eql-mapper/src/model/table_resolver.rs index e461ec488..e6bde466f 100644 --- a/packages/eql-mapper/src/model/table_resolver.rs +++ b/packages/eql-mapper/src/model/table_resolver.rs @@ -1,13 +1,21 @@ use std::sync::{Arc, RwLock}; -use sqltk::parser::ast::{Ident, ObjectName}; +use sqltk::parser::ast::{Ident, ObjectName, ObjectNamePart}; -use super::{Schema, SchemaError, SchemaTableColumn, SchemaWithEdits, Table}; +use super::{IdentCase, Schema, SchemaError, SchemaTableColumn, SchemaWithEdits, Table}; + +/// The schema PostgreSQL keeps a connection's temporary tables in. +const TEMPORARY_SCHEMA: &str = "pg_temp"; #[derive(Debug)] pub enum TableResolver { ViaSchema(Arc), ViaSchemaWithEdits(Arc>), + /// A connection's temporary tables in front of its permanent ones. + WithTemporary { + temporary: Arc, + permanent: Arc, + }, } impl TableResolver { @@ -19,12 +27,32 @@ impl TableResolver { Self::ViaSchemaWithEdits(Arc::new(RwLock::new(SchemaWithEdits::new(schema)))) } + /// An editable resolver for a connection's temporary tables, which DDL + /// fills in. A table in it resolves by its bare name or as `pg_temp.name`. + pub fn new_temporary() -> Self { + Self::new_editable(Arc::new(Schema::new(TEMPORARY_SCHEMA))) + } + + /// Resolves through `temporary` before `permanent`, as PostgreSQL searches + /// `pg_temp` first. + /// + /// A bare name that both resolve does not resolve. PostgreSQL would pick + /// the temporary table, but if Proxy's record of it is stale the name means + /// the permanent table, so neither guess is safe. + pub fn with_temporary(temporary: Arc, permanent: Arc) -> Self { + Self::WithTemporary { + temporary, + permanent, + } + } + pub fn has_schema_changed(&self) -> bool { match self { TableResolver::ViaSchema(_) => false, TableResolver::ViaSchemaWithEdits(schema_with_edits) => { schema_with_edits.read().unwrap().has_schema_changed() } + TableResolver::WithTemporary { permanent, .. } => permanent.has_schema_changed(), } } @@ -32,15 +60,46 @@ impl TableResolver { match self { TableResolver::ViaSchema(_) => None, TableResolver::ViaSchemaWithEdits(schema_with_edits) => Some(schema_with_edits.clone()), + TableResolver::WithTemporary { permanent, .. } => permanent.as_schema_with_edits(), + } + } + + /// The resolver that owns `name`: this one, or for [`Self::WithTemporary`] + /// the layer the name resolves in. + fn owner(&self, name: &ObjectName) -> Result<&TableResolver, SchemaError> { + let TableResolver::WithTemporary { + temporary, + permanent, + } = self + else { + return Ok(self); + }; + + match name.0.as_slice() { + [ObjectNamePart::Identifier(qualifier), _] + if IdentCase(qualifier) == IdentCase(&Ident::new(TEMPORARY_SCHEMA)) => + { + temporary.owner(name) + } + [_] => match ( + temporary.resolve_table(name).is_ok(), + permanent.resolve_table(name).is_ok(), + ) { + (true, true) => Err(SchemaError::AmbiguousTable(name.to_string())), + (true, false) => temporary.owner(name), + (false, _) => permanent.owner(name), + }, + _ => permanent.owner(name), } } pub fn resolve_table(&self, name: &ObjectName) -> Result, SchemaError> { - match self { + match self.owner(name)? { TableResolver::ViaSchema(schema) => schema.resolve_table(name), TableResolver::ViaSchemaWithEdits(schema_with_edits) => { schema_with_edits.read().unwrap().resolve_table(name) } + TableResolver::WithTemporary { .. } => unreachable!("owner is a single layer"), } } @@ -48,12 +107,13 @@ impl TableResolver { &self, table_name: &ObjectName, ) -> Result, SchemaError> { - match self { + match self.owner(table_name)? { TableResolver::ViaSchema(schema) => schema.resolve_table_columns(table_name), TableResolver::ViaSchemaWithEdits(schema_with_edits) => schema_with_edits .read() .unwrap() .resolve_table_columns(table_name), + TableResolver::WithTemporary { .. } => unreachable!("owner is a single layer"), } } @@ -62,7 +122,7 @@ impl TableResolver { table_name: &ObjectName, column_name: &Ident, ) -> Result { - match self { + match self.owner(table_name)? { TableResolver::ViaSchema(schema) => { schema.resolve_table_column(table_name, column_name) } @@ -70,6 +130,107 @@ impl TableResolver { .read() .unwrap() .resolve_table_column(table_name, column_name), + TableResolver::WithTemporary { .. } => unreachable!("owner is a single layer"), } } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use sqltk::parser::ast::{Ident, ObjectName, ObjectNamePart}; + + use crate::{ + collect_ddl, schema, + test_helpers::{id, object_name, parse}, + SchemaError, TableResolver, + }; + + fn qualified(schema_name: &str, table_name: &str) -> ObjectName { + ObjectName(vec![ + ObjectNamePart::Identifier(Ident::new(schema_name)), + ObjectNamePart::Identifier(Ident::new(table_name)), + ]) + } + + /// A resolver over `users (id, email EQL)` with temporary tables created by `ddl`. + fn resolver(ddl: &str) -> TableResolver { + let permanent = Arc::new(TableResolver::new_editable(Arc::new(schema! { + tables: { + users: { + id, + email (EQL), + } + } + }))); + let temporary = Arc::new(TableResolver::new_temporary()); + collect_ddl(temporary.clone(), &parse(ddl)); + TableResolver::with_temporary(temporary, permanent) + } + + #[test] + fn a_bare_name_resolves_to_a_temporary_table() { + let resolver = resolver("create temporary table names (name text)"); + + assert!(resolver + .resolve_table_column(&object_name("names"), &id("name")) + .is_ok()); + } + + #[test] + fn only_a_pg_temp_qualifier_resolves_to_a_temporary_table() { + let resolver = resolver("create temporary table names (name text)"); + + assert!(resolver + .resolve_table(&qualified("pg_temp", "names")) + .is_ok()); + assert_eq!( + resolver.resolve_table(&qualified("public", "names")), + Err(SchemaError::TableNotFound("public.names".into())) + ); + } + + #[test] + fn permanent_tables_resolve_as_before() { + let resolver = resolver("create temporary table names (name text)"); + + assert!(resolver.resolve_table(&object_name("users")).is_ok()); + assert!(resolver + .resolve_table(&qualified("public", "users")) + .is_ok()); + assert!(resolver + .resolve_table(&qualified("pg_temp", "users")) + .is_err()); + } + + /// PostgreSQL would resolve the bare name to the temporary table, but a + /// temporary table Proxy thinks exists may already be gone, and then the + /// name means the permanent table. Neither guess is safe. + #[test] + fn a_temporary_table_that_shadows_a_permanent_one_does_not_resolve() { + let resolver = resolver("create temporary table users (id int)"); + + assert_eq!( + resolver.resolve_table(&object_name("users")), + Err(SchemaError::AmbiguousTable("users".into())) + ); + assert!(resolver + .resolve_table(&qualified("pg_temp", "users")) + .is_ok()); + assert!(resolver + .resolve_table(&qualified("public", "users")) + .is_ok()); + } + + #[test] + fn a_dropped_temporary_table_does_not_resolve() { + let resolver = resolver("create temporary table names (name text)"); + let TableResolver::WithTemporary { temporary, .. } = &resolver else { + unreachable!() + }; + collect_ddl(temporary.clone(), &parse("drop table names")); + + assert!(resolver.resolve_table(&object_name("names")).is_err()); + } +} diff --git a/packages/eql-mapper/src/scope_tracker.rs b/packages/eql-mapper/src/scope_tracker.rs index bc4ffe5e1..631741a70 100644 --- a/packages/eql-mapper/src/scope_tracker.rs +++ b/packages/eql-mapper/src/scope_tracker.rs @@ -2,7 +2,7 @@ use crate::inference::unifier::{ProjectionColumn, Type}; use crate::inference::TypeError; use crate::iterator_ext::IteratorExt; -use crate::model::IdentCase; +use crate::model::{IdentCase, UnresolvedTable}; use crate::unifier::{Projection, Value}; use crate::Relation; use sqltk::parser::ast::{Ident, ObjectName, ObjectNamePart, Query, Statement}; @@ -349,10 +349,18 @@ pub enum ScopeError { } impl ScopeError { - /// See [`TypeError::as_table_not_found`]. - pub fn as_table_not_found(&self) -> Option<&str> { + /// See [`TypeError::is_unsatisfied_bounds`]. + pub fn is_unsatisfied_bounds(&self) -> bool { match self { - ScopeError::TypeError(err) => err.as_table_not_found(), + ScopeError::TypeError(err) => err.is_unsatisfied_bounds(), + _ => false, + } + } + + /// See [`TypeError::as_unresolved_table`]. + pub fn as_unresolved_table(&self) -> Option> { + match self { + ScopeError::TypeError(err) => err.as_unresolved_table(), _ => None, } } diff --git a/tests/python/tests/test_error_messages.py b/tests/python/tests/test_error_messages.py index c0acf3ac0..d3bf67f44 100644 --- a/tests/python/tests/test_error_messages.py +++ b/tests/python/tests/test_error_messages.py @@ -37,7 +37,9 @@ def test_encrypted_column_not_defined_in_schema(): sql = "INSERT INTO encrypted (id, encrypted_unconfigured) VALUES (%s, %s)" - with pytest.raises(psycopg.Error, match=r'relation ".*" does not exist'): + # EQL Mapper cannot type check a write to a column it does not + # know, so Proxy refuses it rather than forward the values as written. + with pytest.raises(psycopg.Error, match='#mapping-statement-could-not-be-type-checked'): cursor.execute(sql, [id, val])