diff --git a/Cargo.lock b/Cargo.lock index 7a89e7d..c4f4ac4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -148,6 +148,16 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cc" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "375989decec1a0fa424442bed520fa12ede43a48475ce8c7417d2c69b32b5f6b" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -194,6 +204,12 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" +[[package]] +name = "deranged" +version = "0.5.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e9de72ce2ad1f90dc62fa25f0f430ef85eb4b0d8fa0be4f30373bc40a21d28e" + [[package]] name = "devcloud-applicationautoscaling" version = "0.1.0" @@ -389,10 +405,13 @@ dependencies = [ "hex", "hmac", "md-5", + "rcgen", + "rustls-pki-types", "serde", "serde_json", "sha2", "tokio", + "tokio-rustls", ] [[package]] @@ -498,6 +517,12 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "find-msvc-tools" +version = "0.1.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48" + [[package]] name = "fixedbitset" version = "0.5.7" @@ -929,12 +954,28 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d" +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64", + "serde_core", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -977,6 +1018,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "powerfmt" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a6394b9e965e73d0a289ee54f589087e2c676aedf60885baf52c76b771e4958" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1102,6 +1149,19 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rcgen" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "yasna", +] + [[package]] name = "regex" version = "1.12.4" @@ -1131,6 +1191,20 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rustix" version = "1.1.4" @@ -1144,6 +1218,40 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.23.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -1227,6 +1335,12 @@ dependencies = [ "digest", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -1337,6 +1451,25 @@ dependencies = [ "syn", ] +[[package]] +name = "time" +version = "0.3.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + [[package]] name = "tokio" version = "1.52.3" @@ -1364,6 +1497,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.26.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -1564,6 +1707,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "utf-8" version = "0.7.6" @@ -1825,6 +1974,15 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "yasna" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd" +dependencies = [ + "time", +] + [[package]] name = "zerocopy" version = "0.8.52" @@ -1845,6 +2003,12 @@ dependencies = [ "syn", ] +[[package]] +name = "zeroize" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13084392c5e4bc371903e2935a5eaeed24905a7511356b883835e18a78f6879" + [[package]] name = "zmij" version = "1.0.21" diff --git a/README.md b/README.md index 2f7a4dc..0610166 100644 --- a/README.md +++ b/README.md @@ -446,8 +446,9 @@ Pub/Sub dashboard actions are available under `/dashboard/pubsub`: | --- | --- | --- | | PostgreSQL wire endpoint | Yes | Listens on `127.0.0.1:15439` by default for local Redshift-style SQL clients. | | PostgreSQL simple query protocol | Yes | Covers `psql` smoke workflows. | +| TLS on the SQL endpoint | Yes | Off by default. Set `services.redshift.tls.certFile` and `services.redshift.tls.keyFile` (PEM; certificate chain leaf first, key in PKCS#8, PKCS#1, or SEC1; paths relative to where `devcloud up` runs) and clients that send an SSLRequest get TLS 1.2 or 1.3 with that certificate, so `sslmode=verify-full` works when the client trusts the issuing CA and the certificate names the host it connects to. devcloud does not issue certificates. Plain connections are still accepted unless `services.redshift.tls.require: true`. The standalone binary reads `DEVCLOUD_REDSHIFT_TLS_CERT_FILE`, `DEVCLOUD_REDSHIFT_TLS_KEY_FILE`, and `DEVCLOUD_REDSHIFT_TLS_REQUIRE`. | | PostgreSQL extended query protocol | Partial | Supports Parse, Bind, Describe, Execute, Sync, Close, text bind parameters, portal resume, and safe unsupported errors for binary formats. | -| PostgreSQL execution backend | Yes | Default backend is managed local PostgreSQL; explicit memory fallback remains available for development continuity. Each SQL client connection gets its own PostgreSQL session, so temp tables, `SET`s, and transactions stay with that client and end when it disconnects. The SQL endpoint does not offer TLS (`sslmode=disable`/`prefer` clients only). | +| PostgreSQL execution backend | Yes | Default backend is managed local PostgreSQL; explicit memory fallback remains available for development continuity. Each SQL client connection gets its own PostgreSQL session, so temp tables, `SET`s, and transactions stay with that client and end when it disconnects. See TLS below. | | Redshift SQL translation | Partial | Handles local subsets for DDL/DML, Redshift table attributes, CTAS/views/materialized-view metadata, function rewrites, COPY, and UNLOAD. | | COPY from local S3 | Partial | `COPY FROM 's3://bucket/prefix'` loads every object under the prefix from the local S3 store into the PostgreSQL backend, on the client's own session (temp tables and open transactions work): default text format with `DELIMITER`, `CSV` / `FORMAT AS CSV`, `JSON 'auto'` (or `'auto ignorecase'`), `GZIP`, `IGNOREHEADER`, `NULL AS`, `EMPTYASNULL`, `BLANKSASNULL`; an empty field is NULL except in character columns, as on Redshift. `IAM_ROLE` / `CREDENTIALS` / `REGION` are accepted and ignored. Column lists, `MANIFEST`, JSONPaths, other compressions, and columnar formats are refused. The memory backend keeps its own CSV/JSON COPY. | | UNLOAD to local S3 | Partial | Memory backend only; the PostgreSQL backend does not run UNLOAD yet. | diff --git a/orchestrator/src/config.rs b/orchestrator/src/config.rs index eacd34a..a644ca3 100644 --- a/orchestrator/src/config.rs +++ b/orchestrator/src/config.rs @@ -272,6 +272,16 @@ pub struct RedshiftServiceConfig { pub data_api: RedshiftDataApiConfig, pub sql: RedshiftSqlConfig, pub copy_unload: RedshiftCopyUnloadConfig, + pub tls: RedshiftTlsConfig, +} + +/// TLS for the Redshift SQL endpoint: PEM files the user supplies. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct RedshiftTlsConfig { + pub cert_file: String, + pub key_file: String, + /// Refuse SQL clients that do not negotiate TLS. + pub require: bool, } #[derive(Debug, Clone, PartialEq, Eq, Default)] @@ -581,6 +591,7 @@ pub fn default_config() -> Config { enable_local_s3: true, max_input_row_bytes: 4 * 1024 * 1024, }, + tls: RedshiftTlsConfig::default(), }, redis: RedisServiceConfig { enabled: false, @@ -1078,6 +1089,13 @@ pub fn apply_config_value(cfg: &mut Config, path: &[String], value: &str) -> io: "services.redshift.sql.defaultSearchPath" => { cfg.services.redshift.sql.default_search_path = value.to_string() } + "services.redshift.tls.certFile" => { + cfg.services.redshift.tls.cert_file = trim_quotes(value) + } + "services.redshift.tls.keyFile" => cfg.services.redshift.tls.key_file = trim_quotes(value), + "services.redshift.tls.require" => { + cfg.services.redshift.tls.require = parse_bool("services.redshift.tls.require", value)? + } "services.redshift.copyUnload.enableLocalS3" => { cfg.services.redshift.copy_unload.enable_local_s3 = parse_bool("services.redshift.copyUnload.enableLocalS3", value)? diff --git a/orchestrator/src/services/redshift.rs b/orchestrator/src/services/redshift.rs index 8e1eccc..322c5fd 100644 --- a/orchestrator/src/services/redshift.rs +++ b/orchestrator/src/services/redshift.rs @@ -100,6 +100,11 @@ pub async fn run( let sql_addr = format!("127.0.0.1:{}", cfg.server.redshift_port); let api_addr = format!("127.0.0.1:{}", cfg.server.redshift_api_port); + let tls = &cfg.services.redshift.tls; + let sql_tls = + devcloud_redshift::tls::sql_tls_from_files(&tls.cert_file, &tls.key_file, tls.require) + .map_err(|e| format!("redshift: {e}"))?; + let server = Arc::new(Server::new(Config { sql_addr: sql_addr.clone(), api_addr: api_addr.clone(), @@ -125,6 +130,7 @@ pub async fn run( sql_backend, translator, events_enabled: true, + sql_tls, })); let sql_listener = TcpListener::bind(&sql_addr) diff --git a/services/redshift/Cargo.toml b/services/redshift/Cargo.toml index 9ae8ed7..f0d37b6 100644 --- a/services/redshift/Cargo.toml +++ b/services/redshift/Cargo.toml @@ -16,6 +16,10 @@ path = "src/main.rs" devcloud-s3 = { path = "../s3" } # gzip COPY sources (pure-Rust backend; miniz_oxide is already in the tree). flate2 = { version = "1", default-features = false, features = ["rust_backend"] } +# TLS on the SQL endpoint (pure-Rust rustls with the ring provider; no +# OpenSSL or CMake needed). Certificates come from files the user supplies. +tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } +rustls-pki-types = { version = "1", features = ["std"] } md-5 = "0.10" sha2 = "0.10" hmac = "0.12" @@ -25,3 +29,7 @@ serde = { version = "1", features = ["derive"] } # map marshaling. Do NOT enable preserve_order. serde_json = "1" tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "net", "io-util", "signal", "time", "sync"] } + +[dev-dependencies] +# Test-only: issues a throwaway CA and server certificate for the TLS tests. +rcgen = { version = "0.13", default-features = false, features = ["ring", "pem"] } diff --git a/services/redshift/src/lib.rs b/services/redshift/src/lib.rs index 96e47ad..e4bf248 100644 --- a/services/redshift/src/lib.rs +++ b/services/redshift/src/lib.rs @@ -52,6 +52,7 @@ pub mod snapshot; pub mod sql_parse; pub mod sql_predicates; pub mod storage; +pub mod tls; pub mod translator; pub use backend::{SqlBackend, SqlTransaction}; diff --git a/services/redshift/src/main.rs b/services/redshift/src/main.rs index 2c34751..aba6102 100644 --- a/services/redshift/src/main.rs +++ b/services/redshift/src/main.rs @@ -24,6 +24,9 @@ //! DEVCLOUD_REDSHIFT_BACKEND_KIND memory | postgres //! DEVCLOUD_REDSHIFT_BACKEND_MODE managed | external | memory | embedded //! DEVCLOUD_REDSHIFT_DSN postgres DSN (managed: daemon-provided) +//! DEVCLOUD_REDSHIFT_TLS_CERT_FILE PEM certificate chain for SQL TLS (optional) +//! DEVCLOUD_REDSHIFT_TLS_KEY_FILE PEM private key for SQL TLS (optional) +//! DEVCLOUD_REDSHIFT_TLS_REQUIRE true: refuse SQL clients without TLS //! //! The daemon owns the managed-PostgreSQL lifecycle (managed_postgres.rs); this //! binary only connects to the DSN. Successful Data API mutations print @@ -82,6 +85,18 @@ fn main() { Some(Arc::new(FileBucketStore::new(object_store_root))) }; + let sql_tls = match devcloud_redshift::tls::sql_tls_from_files( + &env("DEVCLOUD_REDSHIFT_TLS_CERT_FILE"), + &env("DEVCLOUD_REDSHIFT_TLS_KEY_FILE"), + env("DEVCLOUD_REDSHIFT_TLS_REQUIRE").eq_ignore_ascii_case("true"), + ) { + Ok(tls) => tls, + Err(e) => { + eprintln!("devcloud-redshift: {e}"); + std::process::exit(2); + } + }; + let server = Server::new(Config { sql_addr: sql_addr.clone(), api_addr: api_addr.clone(), @@ -103,6 +118,7 @@ fn main() { sql_backend, translator, events_enabled: true, + sql_tls, }); let runtime = tokio::runtime::Builder::new_multi_thread() diff --git a/services/redshift/src/pgwire.rs b/services/redshift/src/pgwire.rs index ba4a72e..c90610f 100644 --- a/services/redshift/src/pgwire.rs +++ b/services/redshift/src/pgwire.rs @@ -57,21 +57,89 @@ impl Server { } }; let server = Arc::clone(&self); - tokio::spawn(async move { server.handle_sql_conn(conn).await }); + tokio::spawn(async move { server.negotiate_and_handle(conn).await }); + } + } + + /// Answers the pre-startup requests a client may send before its startup + /// message: `SSLRequest` (TLS when configured, else `N`) and + /// `GSSENCRequest` (always `N`). The first 8 bytes of anything else are + /// handed back to the startup reader. + async fn negotiate_and_handle(&self, mut conn: tokio::net::TcpStream) { + let mut head = [0u8; 8]; + loop { + if conn.read_exact(&mut head).await.is_err() { + return; + } + let len = i32::from_be_bytes([head[0], head[1], head[2], head[3]]); + let code = i32::from_be_bytes([head[4], head[5], head[6], head[7]]); + if len == 8 && code == PG_SSL_REQUEST_CODE { + if let Some(tls) = &self.sql_tls { + if conn.write_all(b"S").await.is_err() { + return; + } + // A failed handshake (the client does not trust the + // certificate, or its name does not match) just ends + // the connection. + if let Ok(stream) = tls.acceptor.accept(conn).await { + self.handle_sql_conn_secured(stream, true).await; + } + return; + } + if conn.write_all(b"N").await.is_err() { + return; + } + continue; + } + if len == 8 && code == PG_GSSENC_REQUEST_CODE { + if conn.write_all(b"N").await.is_err() { + return; + } + continue; + } + let replay = Prefixed { + prefix: head, + pos: 0, + inner: conn, + }; + self.handle_sql_conn_secured(replay, false).await; + return; } } /// Mirrors `handleSQLConn`: any read/write error tears the connection down /// silently, exactly like legacy bare `return`s. - pub async fn handle_sql_conn(&self, mut conn: S) { - let _ = self.handle_sql_conn_inner(&mut conn).await; + pub async fn handle_sql_conn(&self, conn: S) { + self.handle_sql_conn_secured(conn, false).await + } + + /// [`Self::handle_sql_conn`] for a stream that is (`tls`) or is not + /// encrypted; when TLS is required, a plain stream is refused. + async fn handle_sql_conn_secured( + &self, + mut conn: S, + tls: bool, + ) { + let _ = self.handle_sql_conn_inner(&mut conn, tls).await; } async fn handle_sql_conn_inner( &self, conn: &mut S, + tls: bool, ) -> std::io::Result<()> { let params = self.read_startup(conn).await?; + if !tls && self.sql_tls.as_ref().is_some_and(|t| t.require) { + write_buffered(conn, |buf| { + write_error_response( + buf, + "28000", + "this devcloud Redshift endpoint requires TLS (connect with sslmode=require or stricter)", + ) + }) + .await?; + return Ok(()); + } write_buffered(conn, |buf| write_auth_cleartext_password(buf)).await?; let password = read_password_message(conn).await?; if !self.password_allowed(&password) { @@ -321,3 +389,55 @@ where fn invalid_data(message: &str) -> std::io::Error { std::io::Error::new(std::io::ErrorKind::InvalidData, message) } + +/// GSSAPI encryption request (`GSSENCRequest`); devcloud always declines. +const PG_GSSENC_REQUEST_CODE: i32 = 80877104; + +/// A stream whose first bytes were already read: they are returned again +/// before the rest of `inner`. +struct Prefixed { + prefix: [u8; 8], + pos: usize, + inner: S, +} + +impl AsyncRead for Prefixed { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + if self.pos < self.prefix.len() { + let n = (self.prefix.len() - self.pos).min(buf.remaining()); + let start = self.pos; + buf.put_slice(&self.prefix[start..start + n]); + self.pos += n; + return std::task::Poll::Ready(Ok(())); + } + std::pin::Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl AsyncWrite for Prefixed { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + data: &[u8], + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_write(cx, data) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_shutdown(cx) + } +} diff --git a/services/redshift/src/server.rs b/services/redshift/src/server.rs index 2925de2..5b86ed3 100644 --- a/services/redshift/src/server.rs +++ b/services/redshift/src/server.rs @@ -53,12 +53,15 @@ pub struct Config { /// Daemon seam only: when true, Data API mutations print `DEVCLOUD_EVENT` /// lines on stdout. legacy tests run with a nil publisher (off). pub events_enabled: bool, + /// TLS for the SQL endpoint; `None` answers every SSLRequest with `N`. + pub sql_tls: Option, } pub struct Server { pub(crate) shared: Arc, backend: Arc, translator: Arc, + pub(crate) sql_tls: Option, } pub struct ServerShared { @@ -142,6 +145,7 @@ impl Server { /// legacy), seeds the `public` schema and the default cluster, normalizes /// restored cluster endpoints, and wires the memory backend unless injected. pub fn new(cfg: Config) -> Server { + let sql_tls = cfg.sql_tls.clone(); let config = SharedConfig { sql_addr: cfg.sql_addr, region: cfg.region, @@ -231,6 +235,7 @@ impl Server { shared, backend, translator, + sql_tls, } } diff --git a/services/redshift/src/tls.rs b/services/redshift/src/tls.rs new file mode 100644 index 0000000..e3244a2 --- /dev/null +++ b/services/redshift/src/tls.rs @@ -0,0 +1,63 @@ +//! TLS for the PostgreSQL wire endpoint. +//! +//! devcloud never issues certificates: the user supplies a PEM certificate +//! chain and private key (for example from a local CA its clients trust), and +//! clients that ask for SSL (`SSLRequest`) get a TLS 1.2+ session with them. + +use std::sync::Arc; + +use rustls_pki_types::pem::PemObject; +use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use tokio_rustls::rustls::{self, ServerConfig}; +use tokio_rustls::TlsAcceptor; + +/// TLS settings for the SQL endpoint. +#[derive(Clone)] +pub struct SqlTls { + pub acceptor: TlsAcceptor, + /// Refuse clients that do not negotiate TLS. + pub require: bool, +} + +/// Builds an acceptor from a PEM certificate chain (leaf first) and a PEM +/// private key (PKCS#8, PKCS#1, or SEC1). +pub fn acceptor_from_pem_files(cert_file: &str, key_file: &str) -> Result { + let certs: Vec> = CertificateDer::pem_file_iter(cert_file) + .map_err(|e| format!("read TLS certificate {cert_file}: {e}"))? + .collect::>() + .map_err(|e| format!("parse TLS certificate {cert_file}: {e}"))?; + if certs.is_empty() { + return Err(format!("TLS certificate {cert_file} holds no certificate")); + } + let key = PrivateKeyDer::from_pem_file(key_file) + .map_err(|e| format!("read TLS private key {key_file}: {e}"))?; + let config = + ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + // rustls offers only TLS 1.2 and 1.3. + .with_safe_default_protocol_versions() + .map_err(|e| format!("TLS protocol versions: {e}"))? + .with_no_client_auth() + .with_single_cert(certs, key) + .map_err(|e| format!("TLS certificate {cert_file} and key {key_file}: {e}"))?; + Ok(TlsAcceptor::from(Arc::new(config))) +} + +/// The SQL endpoint's TLS settings from configuration: both files or +/// neither. `require` without a certificate is a configuration error. +pub fn sql_tls_from_files( + cert_file: &str, + key_file: &str, + require: bool, +) -> Result, String> { + match (cert_file.trim().is_empty(), key_file.trim().is_empty()) { + (true, true) if require => { + Err("redshift TLS is required but no certificate and key are configured".to_string()) + } + (true, true) => Ok(None), + (false, false) => Ok(Some(SqlTls { + acceptor: acceptor_from_pem_files(cert_file.trim(), key_file.trim())?, + require, + })), + _ => Err("redshift TLS needs both a certificate file and a key file".to_string()), + } +} diff --git a/services/redshift/tests/pgwire_tls.rs b/services/redshift/tests/pgwire_tls.rs new file mode 100644 index 0000000..a568d3b --- /dev/null +++ b/services/redshift/tests/pgwire_tls.rs @@ -0,0 +1,219 @@ +//! TLS on the PostgreSQL wire endpoint: SSLRequest negotiation, certificate +//! verification by name, plain connections alongside TLS, and `require`. +//! A throwaway CA and server certificate are issued per test with rcgen. + +use std::path::PathBuf; +use std::sync::Arc; + +use devcloud_redshift::tls::sql_tls_from_files; +use devcloud_redshift::{Config, Server}; +use rcgen::{BasicConstraints, CertificateParams, DnType, IsCa, KeyPair, SanType}; +use rustls_pki_types::{CertificateDer, ServerName}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_rustls::rustls::{self, ClientConfig, RootCertStore}; +use tokio_rustls::TlsConnector; + +const SSL_REQUEST: [u8; 8] = [0, 0, 0, 8, 0x04, 0xd2, 0x16, 0x2f]; + +struct Pki { + dir: PathBuf, + ca: CertificateDer<'static>, +} + +impl Drop for Pki { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.dir); + } +} + +impl Pki { + fn cert(&self) -> String { + self.dir.join("server.pem").to_string_lossy().into_owned() + } + fn key(&self) -> String { + self.dir.join("server.key").to_string_lossy().into_owned() + } +} + +/// A CA, and a server certificate it signs for `localhost` and `127.0.0.1`. +fn issue(tag: &str) -> Pki { + let dir = std::env::temp_dir().join(format!( + "devcloud-redshift-tls-{tag}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + let ca_key = KeyPair::generate().unwrap(); + let mut ca_params = CertificateParams::new(Vec::::new()).unwrap(); + ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + ca_params + .distinguished_name + .push(DnType::CommonName, "devcloud test CA"); + let ca = ca_params.self_signed(&ca_key).unwrap(); + + let server_key = KeyPair::generate().unwrap(); + let mut params = CertificateParams::new(vec!["localhost".to_string()]).unwrap(); + params + .subject_alt_names + .push(SanType::IpAddress("127.0.0.1".parse().unwrap())); + let server = params.signed_by(&server_key, &ca, &ca_key).unwrap(); + std::fs::write(dir.join("server.pem"), server.pem()).unwrap(); + std::fs::write(dir.join("server.key"), server_key.serialize_pem()).unwrap(); + Pki { + dir, + ca: ca.der().clone(), + } +} + +async fn start(sql_tls: Option) -> u16 { + let server = Arc::new(Server::new(Config { + sql_tls, + ..Config::default() + })); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { server.serve_sql(listener).await }); + port +} + +fn connector(ca: &CertificateDer<'static>) -> TlsConnector { + let mut roots = RootCertStore::empty(); + roots.add(ca.clone()).unwrap(); + let config = + ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(roots) + .with_no_client_auth(); + TlsConnector::from(Arc::new(config)) +} + +async fn read_message(s: &mut S) -> (u8, Vec) { + let mut tag = [0u8; 1]; + s.read_exact(&mut tag).await.unwrap(); + let mut len = [0u8; 4]; + s.read_exact(&mut len).await.unwrap(); + let mut payload = vec![0u8; u32::from_be_bytes(len) as usize - 4]; + s.read_exact(&mut payload).await.unwrap(); + (tag[0], payload) +} + +async fn send(s: &mut (impl AsyncWrite + Unpin), tag: u8, payload: &[u8]) { + let mut msg = vec![tag]; + msg.extend(((payload.len() + 4) as u32).to_be_bytes()); + msg.extend(payload); + s.write_all(&msg).await.unwrap(); +} + +/// Startup, cleartext password, `select 1`; returns the first message after +/// the startup (an ErrorResponse when the server refuses the session). +async fn login_and_select_one(s: &mut S) -> Result<(), String> { + let mut body = Vec::new(); + body.extend(196608i32.to_be_bytes()); + body.extend(b"user\0dev\0database\0dev\0\0"); + let mut startup = ((body.len() + 4) as u32).to_be_bytes().to_vec(); + startup.extend(body); + s.write_all(&startup).await.unwrap(); + let (tag, payload) = read_message(s).await; + if tag == b'E' { + return Err(String::from_utf8_lossy(&payload).into_owned()); + } + assert_eq!(tag, b'R', "auth request"); + send(s, b'p', b"dev\0").await; + loop { + if read_message(s).await.0 == b'Z' { + break; + } + } + send(s, b'Q', b"select 1\0").await; + let mut saw_row = false; + loop { + match read_message(s).await { + (b'D', payload) if payload.ends_with(b"1") => saw_row = true, + (b'Z', _) => break, + _ => {} + } + } + assert!(saw_row); + send(s, b'X', &[]).await; + Ok(()) +} + +async fn ssl_request(stream: &mut TcpStream) -> u8 { + stream.write_all(&SSL_REQUEST).await.unwrap(); + let mut answer = [0u8; 1]; + stream.read_exact(&mut answer).await.unwrap(); + answer[0] +} + +#[tokio::test(flavor = "multi_thread")] +async fn tls_session_verified_by_name() { + let pki = issue("verified"); + let port = start(sql_tls_from_files(&pki.cert(), &pki.key(), false).unwrap()).await; + for name in ["localhost", "127.0.0.1"] { + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + assert_eq!(ssl_request(&mut tcp).await, b'S'); + let server_name = ServerName::try_from(name.to_string()).unwrap(); + let mut tls = connector(&pki.ca).connect(server_name, tcp).await.unwrap(); + login_and_select_one(&mut tls).await.unwrap(); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn mismatched_name_or_untrusted_ca_fails_the_handshake() { + let pki = issue("mismatch"); + let port = start(sql_tls_from_files(&pki.cert(), &pki.key(), false).unwrap()).await; + + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + assert_eq!(ssl_request(&mut tcp).await, b'S'); + let wrong_name = ServerName::try_from("host.docker.internal".to_string()).unwrap(); + assert!(connector(&pki.ca).connect(wrong_name, tcp).await.is_err()); + + let other = issue("other-ca"); + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + assert_eq!(ssl_request(&mut tcp).await, b'S'); + let name = ServerName::try_from("localhost".to_string()).unwrap(); + assert!(connector(&other.ca).connect(name, tcp).await.is_err()); +} + +#[tokio::test(flavor = "multi_thread")] +async fn plain_connections_still_work_unless_required() { + let pki = issue("plain"); + let port = start(sql_tls_from_files(&pki.cert(), &pki.key(), false).unwrap()).await; + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + login_and_select_one(&mut tcp).await.unwrap(); + + let port = start(sql_tls_from_files(&pki.cert(), &pki.key(), true).unwrap()).await; + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let refused = login_and_select_one(&mut tcp).await.unwrap_err(); + assert!(refused.contains("requires TLS"), "{refused}"); + // TLS still works when required. + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + assert_eq!(ssl_request(&mut tcp).await, b'S'); + let name = ServerName::try_from("localhost".to_string()).unwrap(); + let mut tls = connector(&pki.ca).connect(name, tcp).await.unwrap(); + login_and_select_one(&mut tls).await.unwrap(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn without_tls_ssl_request_is_declined() { + let port = start(None).await; + let mut tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + assert_eq!(ssl_request(&mut tcp).await, b'N'); + login_and_select_one(&mut tcp).await.unwrap(); +} + +#[test] +fn configuration_errors() { + assert!(sql_tls_from_files("", "", false).unwrap().is_none()); + assert!(sql_tls_from_files("", "", true).is_err()); + assert!(sql_tls_from_files("/x/cert.pem", "", false).is_err()); + let err = sql_tls_from_files("/nonexistent/cert.pem", "/nonexistent/key.pem", false) + .err() + .unwrap(); + assert!(err.contains("/nonexistent/cert.pem"), "{err}"); +}