Rename write_message to write_message_noflush in postgres_backend_async.rs

To make it unifrom across the project; proxy stream.rs and older
postgres_backend uses write_message_noflush.

Dockerfile.vm-compute-node

@@ -10,7 +10,6 @@ RUN set -e \
 	&& rm -f /etc/inittab \
 	&& touch /etc/inittab
 
-ADD vm-cgconfig.conf /etc/cgconfig.conf
 RUN set -e \
 	&& echo "::sysinit:cgconfigparser -l /etc/cgconfig.conf -s 1664" >> /etc/inittab \
 	&& echo "::respawn:su vm-informant -c '/usr/local/bin/vm-informant --auto-restart --cgroup=neon-postgres'" >> /etc/inittab
@@ -26,6 +25,7 @@ RUN apt update && \
 RUN adduser vm-informant --disabled-password --no-create-home
 USER postgres
 
+ADD vm-cgconfig.conf /etc/cgconfig.conf
 COPY --from=informant /etc/inittab /etc/inittab
 COPY --from=informant /usr/bin/vm-informant /usr/local/bin/vm-informant
 

libs/utils/src/postgres_backend_async.rs

@@ -233,7 +233,7 @@ impl PostgresBackend {
     }
 
     /// Write message into internal output buffer.
-    pub fn write_message(&mut self, message: &BeMessage<'_>) -> io::Result<&mut Self> {
+    pub fn write_message_noflush(&mut self, message: &BeMessage<'_>) -> io::Result<&mut Self> {
         BeMessage::write(&mut self.buf_out, message)?;
         Ok(self)
     }
@@ -383,7 +383,7 @@ impl PostgresBackend {
                     FeStartupPacket::SslRequest => {
                         debug!("SSL requested");
 
-                        self.write_message(&BeMessage::EncryptionResponse(have_tls))?;
+                        self.write_message_noflush(&BeMessage::EncryptionResponse(have_tls))?;
                         if have_tls {
                             self.start_tls().await?;
                             self.state = ProtoState::Encrypted;
@@ -391,11 +391,11 @@ impl PostgresBackend {
                     }
                     FeStartupPacket::GssEncRequest => {
                         debug!("GSS requested");
-                        self.write_message(&BeMessage::EncryptionResponse(false))?;
+                        self.write_message_noflush(&BeMessage::EncryptionResponse(false))?;
                     }
                     FeStartupPacket::StartupMessage { .. } => {
                         if have_tls && !matches!(self.state, ProtoState::Encrypted) {
-                            self.write_message(&BeMessage::ErrorResponse(
+                            self.write_message_noflush(&BeMessage::ErrorResponse(
                                 "must connect with TLS",
                                 None,
                             ))?;
@@ -410,15 +410,17 @@ impl PostgresBackend {
 
                         match self.auth_type {
                             AuthType::Trust => {
-                                self.write_message(&BeMessage::AuthenticationOk)?
-                                    .write_message(&BeMessage::CLIENT_ENCODING)?
+                                self.write_message_noflush(&BeMessage::AuthenticationOk)?
+                                    .write_message_noflush(&BeMessage::CLIENT_ENCODING)?
                                     // The async python driver requires a valid server_version
-                                    .write_message(&BeMessage::server_version("14.1"))?
-                                    .write_message(&BeMessage::ReadyForQuery)?;
+                                    .write_message_noflush(&BeMessage::server_version("14.1"))?
+                                    .write_message_noflush(&BeMessage::ReadyForQuery)?;
                                 self.state = ProtoState::Established;
                             }
                             AuthType::NeonJWT => {
-                                self.write_message(&BeMessage::AuthenticationCleartextPassword)?;
+                                self.write_message_noflush(
+                                    &BeMessage::AuthenticationCleartextPassword,
+                                )?;
                                 self.state = ProtoState::Authentication;
                             }
                         }
@@ -441,7 +443,7 @@ impl PostgresBackend {
                         let (_, jwt_response) = m.split_last().context("protocol violation")?;
 
                         if let Err(e) = handler.check_auth_jwt(self, jwt_response) {
-                            self.write_message(&BeMessage::ErrorResponse(
+                            self.write_message_noflush(&BeMessage::ErrorResponse(
                                 &e.to_string(),
                                 Some(e.pg_error_code()),
                             ))?;
@@ -449,9 +451,9 @@ impl PostgresBackend {
                         }
                     }
                 }
-                self.write_message(&BeMessage::AuthenticationOk)?
-                    .write_message(&BeMessage::CLIENT_ENCODING)?
-                    .write_message(&BeMessage::ReadyForQuery)?;
+                self.write_message_noflush(&BeMessage::AuthenticationOk)?
+                    .write_message_noflush(&BeMessage::CLIENT_ENCODING)?
+                    .write_message_noflush(&BeMessage::ReadyForQuery)?;
                 self.state = ProtoState::Established;
             }
 
@@ -486,30 +488,30 @@ impl PostgresBackend {
                 if let Err(e) = handler.process_query(self, query_string).await {
                     log_query_error(query_string, &e);
                     let short_error = short_error(&e);
-                    self.write_message(&BeMessage::ErrorResponse(
+                    self.write_message_noflush(&BeMessage::ErrorResponse(
                         &short_error,
                         Some(e.pg_error_code()),
                     ))?;
                 }
-                self.write_message(&BeMessage::ReadyForQuery)?;
+                self.write_message_noflush(&BeMessage::ReadyForQuery)?;
             }
 
             FeMessage::Parse(m) => {
                 *unnamed_query_string = m.query_string;
-                self.write_message(&BeMessage::ParseComplete)?;
+                self.write_message_noflush(&BeMessage::ParseComplete)?;
             }
 
             FeMessage::Describe(_) => {
-                self.write_message(&BeMessage::ParameterDescription)?
-                    .write_message(&BeMessage::NoData)?;
+                self.write_message_noflush(&BeMessage::ParameterDescription)?
+                    .write_message_noflush(&BeMessage::NoData)?;
             }
 
             FeMessage::Bind(_) => {
-                self.write_message(&BeMessage::BindComplete)?;
+                self.write_message_noflush(&BeMessage::BindComplete)?;
             }
 
             FeMessage::Close(_) => {
-                self.write_message(&BeMessage::CloseComplete)?;
+                self.write_message_noflush(&BeMessage::CloseComplete)?;
             }
 
             FeMessage::Execute(_) => {
@@ -517,7 +519,7 @@ impl PostgresBackend {
                 trace!("got execute {query_string:?}");
                 if let Err(e) = handler.process_query(self, query_string).await {
                     log_query_error(query_string, &e);
-                    self.write_message(&BeMessage::ErrorResponse(
+                    self.write_message_noflush(&BeMessage::ErrorResponse(
                         &e.to_string(),
                         Some(e.pg_error_code()),
                     ))?;
@@ -529,7 +531,7 @@ impl PostgresBackend {
             }
 
             FeMessage::Sync => {
-                self.write_message(&BeMessage::ReadyForQuery)?;
+                self.write_message_noflush(&BeMessage::ReadyForQuery)?;
             }
 
             FeMessage::Terminate => {
@@ -579,7 +581,7 @@ impl<'a> AsyncWrite for CopyDataWriter<'a> {
         // XXX: if the input is large, we should split it into multiple messages.
         // Not sure what the threshold should be, but the ultimate hard limit is that
         // the length cannot exceed u32.
-        this.pgb.write_message(&BeMessage::CopyData(buf))?;
+        this.pgb.write_message_noflush(&BeMessage::CopyData(buf))?;
 
         Poll::Ready(Ok(buf.len()))
     }

pageserver/src/page_service.rs

@@ -64,7 +64,7 @@ fn copyin_stream(pgb: &mut PostgresBackend) -> impl Stream<Item = io::Result<Byt
                 _ = task_mgr::shutdown_watcher() => {
                     // We were requested to shut down.
                     let msg = format!("pageserver is shutting down");
-                    let _ = pgb.write_message(&BeMessage::ErrorResponse(&msg, None));
+                    let _ = pgb.write_message_noflush(&BeMessage::ErrorResponse(&msg, None));
                     Err(QueryError::Other(anyhow::anyhow!(msg)))
                 }
 
@@ -80,13 +80,13 @@ fn copyin_stream(pgb: &mut PostgresBackend) -> impl Stream<Item = io::Result<Byt
                         FeMessage::Terminate => {
                             let msg = "client terminated connection with Terminate message during COPY";
                             let query_error_error = QueryError::Disconnected(ConnectionError::Socket(io::Error::new(io::ErrorKind::ConnectionReset, msg)));
-                            pgb.write_message(&BeMessage::ErrorResponse(msg, Some(query_error_error.pg_error_code())))?;
+                            pgb.write_message_noflush(&BeMessage::ErrorResponse(msg, Some(query_error_error.pg_error_code())))?;
                             Err(io::Error::new(io::ErrorKind::ConnectionReset, msg))?;
                             break;
                         }
                         m => {
                             let msg = format!("unexpected message {m:?}");
-                            pgb.write_message(&BeMessage::ErrorResponse(&msg, None))?;
+                            pgb.write_message_noflush(&BeMessage::ErrorResponse(&msg, None))?;
                             Err(io::Error::new(io::ErrorKind::Other, msg))?;
                             break;
                         }
@@ -97,7 +97,7 @@ fn copyin_stream(pgb: &mut PostgresBackend) -> impl Stream<Item = io::Result<Byt
                 Ok(None) => {
                     let msg = "client closed connection during COPY";
                     let query_error_error = QueryError::Disconnected(ConnectionError::Socket(io::Error::new(io::ErrorKind::ConnectionReset, msg)));
-                    pgb.write_message(&BeMessage::ErrorResponse(msg, Some(query_error_error.pg_error_code())))?;
+                    pgb.write_message_noflush(&BeMessage::ErrorResponse(msg, Some(query_error_error.pg_error_code())))?;
                     pgb.flush().await?;
                     Err(io::Error::new(io::ErrorKind::ConnectionReset, msg))?;
                 }
@@ -311,7 +311,7 @@ impl PageServerHandler {
         let timeline = tenant.get_timeline(timeline_id, true)?;
 
         // switch client to COPYBOTH
-        pgb.write_message(&BeMessage::CopyBothResponse)?;
+        pgb.write_message_noflush(&BeMessage::CopyBothResponse)?;
         pgb.flush().await?;
 
         let metrics = PageRequestMetrics::new(&tenant_id, &timeline_id);
@@ -380,7 +380,7 @@ impl PageServerHandler {
                 })
             });
 
-            pgb.write_message(&BeMessage::CopyData(&response.serialize()))?;
+            pgb.write_message_noflush(&BeMessage::CopyData(&response.serialize()))?;
             pgb.flush().await?;
         }
         Ok(())
@@ -416,7 +416,7 @@ impl PageServerHandler {
 
         // Import basebackup provided via CopyData
         info!("importing basebackup");
-        pgb.write_message(&BeMessage::CopyInResponse)?;
+        pgb.write_message_noflush(&BeMessage::CopyInResponse)?;
         pgb.flush().await?;
 
         let mut copyin_stream = Box::pin(copyin_stream(pgb));
@@ -468,7 +468,7 @@ impl PageServerHandler {
 
         // Import wal provided via CopyData
         info!("importing wal");
-        pgb.write_message(&BeMessage::CopyInResponse)?;
+        pgb.write_message_noflush(&BeMessage::CopyInResponse)?;
         pgb.flush().await?;
         let mut copyin_stream = Box::pin(copyin_stream(pgb));
         let mut reader = tokio_util::io::StreamReader::new(&mut copyin_stream);
@@ -678,7 +678,7 @@ impl PageServerHandler {
         }
 
         // switch client to COPYOUT
-        pgb.write_message(&BeMessage::CopyOutResponse)?;
+        pgb.write_message_noflush(&BeMessage::CopyOutResponse)?;
         pgb.flush().await?;
 
         // Send a tarball of the latest layer on the timeline
@@ -695,7 +695,7 @@ impl PageServerHandler {
             .await?;
         }
 
-        pgb.write_message(&BeMessage::CopyDone)?;
+        pgb.write_message_noflush(&BeMessage::CopyDone)?;
         pgb.flush().await?;
         info!("basebackup complete");
 
@@ -812,7 +812,7 @@ impl postgres_backend_async::Handler for PageServerHandler {
             // Check that the timeline exists
             self.handle_basebackup_request(pgb, tenant_id, timeline_id, lsn, None, false, ctx)
                 .await?;
-            pgb.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
         }
         // return pair of prev_lsn and last_lsn
         else if query_string.starts_with("get_last_record_rlsn ") {
@@ -835,15 +835,15 @@ impl postgres_backend_async::Handler for PageServerHandler {
 
             let end_of_timeline = timeline.get_last_record_rlsn();
 
-            pgb.write_message(&BeMessage::RowDescription(&[
+            pgb.write_message_noflush(&BeMessage::RowDescription(&[
                 RowDescriptor::text_col(b"prev_lsn"),
                 RowDescriptor::text_col(b"last_lsn"),
             ]))?
-            .write_message(&BeMessage::DataRow(&[
+            .write_message_noflush(&BeMessage::DataRow(&[
                 Some(end_of_timeline.prev.to_string().as_bytes()),
                 Some(end_of_timeline.last.to_string().as_bytes()),
             ]))?
-            .write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            .write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
         }
         // same as basebackup, but result includes relational data as well
         else if query_string.starts_with("fullbackup ") {
@@ -884,7 +884,7 @@ impl postgres_backend_async::Handler for PageServerHandler {
             // Check that the timeline exists
             self.handle_basebackup_request(pgb, tenant_id, timeline_id, lsn, prev_lsn, true, ctx)
                 .await?;
-            pgb.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
         } else if query_string.starts_with("import basebackup ") {
             // Import the `base` section (everything but the wal) of a basebackup.
             // Assumes the tenant already exists on this pageserver.
@@ -929,10 +929,10 @@ impl postgres_backend_async::Handler for PageServerHandler {
                 )
                 .await
             {
-                Ok(()) => pgb.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?,
+                Ok(()) => pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?,
                 Err(e) => {
                     error!("error importing base backup between {base_lsn} and {end_lsn}: {e:?}");
-                    pgb.write_message(&BeMessage::ErrorResponse(
+                    pgb.write_message_noflush(&BeMessage::ErrorResponse(
                         &e.to_string(),
                         Some(e.pg_error_code()),
                     ))?
@@ -965,10 +965,10 @@ impl postgres_backend_async::Handler for PageServerHandler {
                 .handle_import_wal(pgb, tenant_id, timeline_id, start_lsn, end_lsn, ctx)
                 .await
             {
-                Ok(()) => pgb.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?,
+                Ok(()) => pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?,
                 Err(e) => {
                     error!("error importing WAL between {start_lsn} and {end_lsn}: {e:?}");
-                    pgb.write_message(&BeMessage::ErrorResponse(
+                    pgb.write_message_noflush(&BeMessage::ErrorResponse(
                         &e.to_string(),
                         Some(e.pg_error_code()),
                     ))?
@@ -977,7 +977,7 @@ impl postgres_backend_async::Handler for PageServerHandler {
         } else if query_string.to_ascii_lowercase().starts_with("set ") {
             // important because psycopg2 executes "SET datestyle TO 'ISO'"
             // on connect
-            pgb.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            pgb.write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
         } else if query_string.starts_with("show ") {
             // show <tenant_id>
             let (_, params_raw) = query_string.split_at("show ".len());
@@ -993,7 +993,7 @@ impl postgres_backend_async::Handler for PageServerHandler {
             self.check_permission(Some(tenant_id))?;
 
             let tenant = get_active_tenant_with_timeout(tenant_id, &ctx).await?;
-            pgb.write_message(&BeMessage::RowDescription(&[
+            pgb.write_message_noflush(&BeMessage::RowDescription(&[
                 RowDescriptor::int8_col(b"checkpoint_distance"),
                 RowDescriptor::int8_col(b"checkpoint_timeout"),
                 RowDescriptor::int8_col(b"compaction_target_size"),
@@ -1004,7 +1004,7 @@ impl postgres_backend_async::Handler for PageServerHandler {
                 RowDescriptor::int8_col(b"image_creation_threshold"),
                 RowDescriptor::int8_col(b"pitr_interval"),
             ]))?
-            .write_message(&BeMessage::DataRow(&[
+            .write_message_noflush(&BeMessage::DataRow(&[
                 Some(tenant.get_checkpoint_distance().to_string().as_bytes()),
                 Some(
                     tenant
@@ -1027,7 +1027,7 @@ impl postgres_backend_async::Handler for PageServerHandler {
                 Some(tenant.get_image_creation_threshold().to_string().as_bytes()),
                 Some(tenant.get_pitr_interval().as_secs().to_string().as_bytes()),
             ]))?
-            .write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;
+            .write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
         } else {
             return Err(QueryError::Other(anyhow::anyhow!(
                 "unknown command {query_string}"

proxy/src/auth.rs

@@ -3,7 +3,7 @@
 pub mod backend;
 pub use backend::BackendType;
 
-pub mod credentials;
+mod credentials;
 pub use credentials::ClientCredentials;
 
 mod password_hack;

proxy/src/auth/backend.rs

@@ -11,7 +11,7 @@ use crate::{
         provider::{CachedNodeInfo, ConsoleReqExtra},
         Api,
     },
-    scram, stream, url,
+    stream, url,
 };
 use futures::TryFutureExt;
 use std::borrow::Cow;
@@ -59,8 +59,8 @@ impl std::fmt::Display for BackendType<'_, ()> {
     fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         use BackendType::*;
         match self {
-            Console(api, _) => fmt.debug_tuple("Console").field(&api.url()).finish(),
-            Postgres(api, _) => fmt.debug_tuple("Postgres").field(&api.url()).finish(),
+            Console(endpoint, _) => fmt.debug_tuple("Console").field(&endpoint.url()).finish(),
+            Postgres(endpoint, _) => fmt.debug_tuple("Postgres").field(&endpoint.url()).finish(),
             Link(url) => fmt.debug_tuple("Link").field(&url.as_str()).finish(),
         }
     }
@@ -106,23 +106,6 @@ impl<'a, T, E> BackendType<'a, Result<T, E>> {
     }
 }
 
-impl console::AuthInfo {
-    /// Either it's our way ([SCRAM](crate::scram)) or the highway :)
-    /// But seriously, we don't aim to support anything but SCRAM for now.
-    fn scram_or_goodbye(self) -> auth::Result<scram::ServerSecret> {
-        match self {
-            Self::Md5(_) => {
-                info!("auth endpoint chooses MD5");
-                Err(auth::AuthError::bad_auth_method("MD5"))
-            }
-            Self::Scram(secret) => {
-                info!("auth endpoint chooses SCRAM");
-                Ok(secret)
-            }
-        }
-    }
-}
-
 /// True to its name, this function encapsulates our current auth trade-offs.
 /// Here, we choose the appropriate auth flow based on circumstances.
 async fn auth_quirks(
@@ -200,9 +183,7 @@ impl BackendType<'_, ClientCredentials<'_>> {
         info!("user successfully authenticated");
         Ok(res)
     }
-}
 
-impl BackendType<'_, ClientCredentials<'_>> {
     /// When applicable, wake the compute node, gaining its connection info in the process.
     /// The link auth flow doesn't support this, so we return [`None`] in that case.
     pub async fn wake_compute(

proxy/src/auth/backend/classic.rs

@@ -2,54 +2,57 @@ use super::AuthSuccess;
 use crate::{
     auth::{self, AuthFlow, ClientCredentials},
     compute,
-    console::{self, CachedNodeInfo, ConsoleReqExtra},
+    console::{self, AuthInfo, CachedNodeInfo, ConsoleReqExtra},
     sasl, scram,
     stream::PqStream,
 };
 use tokio::io::{AsyncRead, AsyncWrite};
-use tokio_postgres::config::AuthKeys;
 use tracing::info;
 
-async fn do_scram(
-    secret: scram::ServerSecret,
-    creds: &ClientCredentials<'_>,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-) -> auth::Result<compute::ScramKeys> {
-    let outcome = AuthFlow::new(client)
-        .begin(auth::Scram(&secret))
-        .await?
-        .authenticate()
-        .await?;
-
-    let client_key = match outcome {
-        sasl::Outcome::Success(key) => key,
-        sasl::Outcome::Failure(reason) => {
-            info!("auth backend failed with an error: {reason}");
-            return Err(auth::AuthError::auth_failed(creds.user));
-        }
-    };
-
-    let keys = compute::ScramKeys {
-        client_key: client_key.as_bytes(),
-        server_key: secret.server_key.as_bytes(),
-    };
-
-    Ok(keys)
-}
-
-pub async fn authenticate(
+pub(super) async fn authenticate(
     api: &impl console::Api,
     extra: &ConsoleReqExtra<'_>,
     creds: &ClientCredentials<'_>,
     client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
 ) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
-    let info = console::get_auth_info(api, extra, creds).await?;
+    info!("fetching user's authentication info");
+    let info = api.get_auth_info(extra, creds).await?.unwrap_or_else(|| {
+        // If we don't have an authentication secret, we mock one to
+        // prevent malicious probing (possible due to missing protocol steps).
+        // This mocked secret will never lead to successful authentication.
+        info!("authentication info not found, mocking it");
+        AuthInfo::Scram(scram::ServerSecret::mock(creds.user, rand::random()))
+    });
 
-    let secret = info.scram_or_goodbye()?;
-    let scram_keys = do_scram(secret, creds, client).await?;
+    let flow = AuthFlow::new(client);
+    let scram_keys = match info {
+        AuthInfo::Md5(_) => {
+            info!("auth endpoint chooses MD5");
+            return Err(auth::AuthError::bad_auth_method("MD5"));
+        }
+        AuthInfo::Scram(secret) => {
+            info!("auth endpoint chooses SCRAM");
+            let scram = auth::Scram(&secret);
+            let client_key = match flow.begin(scram).await?.authenticate().await? {
+                sasl::Outcome::Success(key) => key,
+                sasl::Outcome::Failure(reason) => {
+                    info!("auth backend failed with an error: {reason}");
+                    return Err(auth::AuthError::auth_failed(creds.user));
+                }
+            };
+
+            Some(compute::ScramKeys {
+                client_key: client_key.as_bytes(),
+                server_key: secret.server_key.as_bytes(),
+            })
+        }
+    };
 
     let mut node = api.wake_compute(extra, creds).await?;
-    node.config.auth_keys(AuthKeys::ScramSha256(scram_keys));
+    if let Some(keys) = scram_keys {
+        use tokio_postgres::config::AuthKeys;
+        node.config.auth_keys(AuthKeys::ScramSha256(keys));
+    }
 
     Ok(AuthSuccess {
         reported_auth_ok: false,

proxy/src/auth/backend/hacks.rs

@@ -5,33 +5,11 @@ use crate::{
         self,
         provider::{CachedNodeInfo, ConsoleReqExtra},
     },
-    stream::PqStream,
+    stream,
 };
 use tokio::io::{AsyncRead, AsyncWrite};
 use tracing::{info, warn};
 
-/// Wake the compute node, but only if the password is valid.
-async fn get_compute(
-    api: &impl console::Api,
-    extra: &ConsoleReqExtra<'_>,
-    creds: &mut ClientCredentials<'_>,
-    password: Vec<u8>,
-) -> auth::Result<CachedNodeInfo> {
-    // TODO: this will slow down both "hacks" below; we probably need a cache.
-    let info = console::get_auth_info(api, extra, creds).await?;
-
-    let secret = info.scram_or_goodbye()?;
-    if !secret.matches_password(&password) {
-        info!("our obscure magic indicates that the password doesn't match");
-        return Err(auth::AuthError::auth_failed(creds.user));
-    }
-
-    let mut node = api.wake_compute(extra, creds).await?;
-    node.config.password(password);
-
-    Ok(node)
-}
-
 /// Compared to [SCRAM](crate::scram), cleartext password auth saves
 /// one round trip and *expensive* computations (>= 4096 HMAC iterations).
 /// These properties are benefical for serverless JS workers, so we
@@ -40,7 +18,7 @@ pub async fn cleartext_hack(
     api: &impl console::Api,
     extra: &ConsoleReqExtra<'_>,
     creds: &mut ClientCredentials<'_>,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
+    client: &mut stream::PqStream<impl AsyncRead + AsyncWrite + Unpin>,
 ) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
     warn!("cleartext auth flow override is enabled, proceeding");
     let password = AuthFlow::new(client)
@@ -49,7 +27,8 @@ pub async fn cleartext_hack(
         .authenticate()
         .await?;
 
-    let node = get_compute(api, extra, creds, password).await?;
+    let mut node = api.wake_compute(extra, creds).await?;
+    node.config.password(password);
 
     // Report tentative success; compute node will check the password anyway.
     Ok(AuthSuccess {
@@ -64,7 +43,7 @@ pub async fn password_hack(
     api: &impl console::Api,
     extra: &ConsoleReqExtra<'_>,
     creds: &mut ClientCredentials<'_>,
-    client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
+    client: &mut stream::PqStream<impl AsyncRead + AsyncWrite + Unpin>,
 ) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
     warn!("project not specified, resorting to the password hack auth flow");
     let payload = AuthFlow::new(client)
@@ -76,7 +55,8 @@ pub async fn password_hack(
     info!(project = &payload.project, "received missing parameter");
     creds.project = Some(payload.project.into());
 
-    let node = get_compute(api, extra, creds, payload.password).await?;
+    let mut node = api.wake_compute(extra, creds).await?;
+    node.config.password(payload.password);
 
     // Report tentative success; compute node will check the password anyway.
     Ok(AuthSuccess {

proxy/src/auth/credentials.rs

@@ -31,22 +31,12 @@ pub enum ClientCredsParseError {
 
 impl UserFacingError for ClientCredsParseError {}
 
-/// eSNI parameters which might contain endpoint/project name.
-#[derive(Default)]
-pub struct SniParams<'a> {
-    /// Server Name Indication (TLS jargon).
-    pub sni: Option<&'a str>,
-    /// Common Name from a TLS certificate.
-    pub common_name: Option<&'a str>,
-}
-
 /// Various client credentials which we use for authentication.
 /// Note that we don't store any kind of client key or password here.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ClientCredentials<'a> {
-    /// Name of postgres role.
     pub user: &'a str,
-    /// Also known as endpoint in the console.
+    // TODO: this is a severe misnomer! We should think of a new name ASAP.
     pub project: Option<Cow<'a, str>>,
 }
 
@@ -59,17 +49,18 @@ impl ClientCredentials<'_> {
 
 impl<'a> ClientCredentials<'a> {
     pub fn parse(
-        startup_params: &'a StartupMessageParams,
-        &SniParams { sni, common_name }: &SniParams<'_>,
+        params: &'a StartupMessageParams,
+        sni: Option<&str>,
+        common_name: Option<&str>,
     ) -> Result<Self, ClientCredsParseError> {
         use ClientCredsParseError::*;
 
         // Some parameters are stored in the startup message.
-        let get_param = |key| startup_params.get(key).ok_or(MissingKey(key));
+        let get_param = |key| params.get(key).ok_or(MissingKey(key));
         let user = get_param("user")?;
 
         // Project name might be passed via PG's command-line options.
-        let project_option = startup_params.options_raw().and_then(|mut options| {
+        let project_option = params.options_raw().and_then(|mut options| {
             options
                 .find_map(|opt| opt.strip_prefix("project="))
                 .map(Cow::Borrowed)
@@ -131,9 +122,7 @@ mod tests {
         // According to postgresql, only `user` should be required.
         let options = StartupMessageParams::new([("user", "john_doe")]);
 
-        let sni = SniParams::default();
-
-        let creds = ClientCredentials::parse(&options, &sni)?;
+        let creds = ClientCredentials::parse(&options, None, None)?;
         assert_eq!(creds.user, "john_doe");
         assert_eq!(creds.project, None);
 
@@ -142,15 +131,13 @@ mod tests {
 
     #[test]
     fn parse_excessive() -> anyhow::Result<()> {
-        let startup = StartupMessageParams::new([
+        let options = StartupMessageParams::new([
             ("user", "john_doe"),
             ("database", "world"), // should be ignored
             ("foo", "bar"),        // should be ignored
         ]);
 
-        let sni = SniParams::default();
-
-        let creds = ClientCredentials::parse(&startup, &sni)?;
+        let creds = ClientCredentials::parse(&options, None, None)?;
         assert_eq!(creds.user, "john_doe");
         assert_eq!(creds.project, None);
 
@@ -159,14 +146,12 @@ mod tests {
 
     #[test]
     fn parse_project_from_sni() -> anyhow::Result<()> {
-        let startup = StartupMessageParams::new([("user", "john_doe")]);
+        let options = StartupMessageParams::new([("user", "john_doe")]);
 
-        let sni = SniParams {
-            sni: Some("foo.localhost"),
-            common_name: Some("localhost"),
-        };
+        let sni = Some("foo.localhost");
+        let common_name = Some("localhost");
 
-        let creds = ClientCredentials::parse(&startup, &sni)?;
+        let creds = ClientCredentials::parse(&options, sni, common_name)?;
         assert_eq!(creds.user, "john_doe");
         assert_eq!(creds.project.as_deref(), Some("foo"));
 
@@ -175,14 +160,12 @@ mod tests {
 
     #[test]
     fn parse_project_from_options() -> anyhow::Result<()> {
-        let startup = StartupMessageParams::new([
+        let options = StartupMessageParams::new([
             ("user", "john_doe"),
             ("options", "-ckey=1 project=bar -c geqo=off"),
         ]);
 
-        let sni = SniParams::default();
-
-        let creds = ClientCredentials::parse(&startup, &sni)?;
+        let creds = ClientCredentials::parse(&options, None, None)?;
         assert_eq!(creds.user, "john_doe");
         assert_eq!(creds.project.as_deref(), Some("bar"));
 
@@ -191,17 +174,12 @@ mod tests {
 
     #[test]
     fn parse_projects_identical() -> anyhow::Result<()> {
-        let startup = StartupMessageParams::new([
-            ("user", "john_doe"),
-            ("options", "project=baz"), // fmt
-        ]);
+        let options = StartupMessageParams::new([("user", "john_doe"), ("options", "project=baz")]);
 
-        let sni = SniParams {
-            sni: Some("baz.localhost"),
-            common_name: Some("localhost"),
-        };
+        let sni = Some("baz.localhost");
+        let common_name = Some("localhost");
 
-        let creds = ClientCredentials::parse(&startup, &sni)?;
+        let creds = ClientCredentials::parse(&options, sni, common_name)?;
         assert_eq!(creds.user, "john_doe");
         assert_eq!(creds.project.as_deref(), Some("baz"));
 
@@ -210,17 +188,13 @@ mod tests {
 
     #[test]
     fn parse_projects_different() {
-        let startup = StartupMessageParams::new([
-            ("user", "john_doe"),
-            ("options", "project=first"), // fmt
-        ]);
+        let options =
+            StartupMessageParams::new([("user", "john_doe"), ("options", "project=first")]);
 
-        let sni = SniParams {
-            sni: Some("second.localhost"),
-            common_name: Some("localhost"),
-        };
+        let sni = Some("second.localhost");
+        let common_name = Some("localhost");
 
-        let err = ClientCredentials::parse(&startup, &sni).expect_err("should fail");
+        let err = ClientCredentials::parse(&options, sni, common_name).expect_err("should fail");
         match err {
             InconsistentProjectNames { domain, option } => {
                 assert_eq!(option, "first");
@@ -232,14 +206,12 @@ mod tests {
 
     #[test]
     fn parse_inconsistent_sni() {
-        let startup = StartupMessageParams::new([("user", "john_doe")]);
+        let options = StartupMessageParams::new([("user", "john_doe")]);
 
-        let sni = SniParams {
-            sni: Some("project.localhost"),
-            common_name: Some("example.com"),
-        };
+        let sni = Some("project.localhost");
+        let common_name = Some("example.com");
 
-        let err = ClientCredentials::parse(&startup, &sni).expect_err("should fail");
+        let err = ClientCredentials::parse(&options, sni, common_name).expect_err("should fail");
         match err {
             InconsistentSni { sni, cn } => {
                 assert_eq!(sni, "project.localhost");

proxy/src/console.rs

@@ -6,9 +6,7 @@ pub mod messages;
 
 /// Wrappers for console APIs and their mocks.
 pub mod provider;
-pub use provider::{
-    errors, get_auth_info, Api, AuthInfo, CachedNodeInfo, ConsoleReqExtra, NodeInfo,
-};
+pub use provider::{errors, Api, AuthInfo, CachedNodeInfo, ConsoleReqExtra, NodeInfo};
 
 /// Various cache-related types.
 pub mod caches {

proxy/src/console/provider.rs

@@ -9,7 +9,6 @@ use crate::{
 };
 use async_trait::async_trait;
 use std::sync::Arc;
-use tracing::info;
 
 pub mod errors {
     use crate::{
@@ -176,12 +175,6 @@ pub struct NodeInfo {
 pub type NodeInfoCache = TimedLru<Arc<str>, NodeInfo>;
 pub type CachedNodeInfo = timed_lru::Cached<&'static NodeInfoCache>;
 
-/// Various caches for [`console`].
-pub struct ApiCaches {
-    /// Cache for the `wake_compute` API method.
-    pub node_info: NodeInfoCache,
-}
-
 /// This will allocate per each call, but the http requests alone
 /// already require a few allocations, so it should be fine.
 #[async_trait]
@@ -201,21 +194,8 @@ pub trait Api {
     ) -> Result<CachedNodeInfo, errors::WakeComputeError>;
 }
 
-/// A more insightful version of [`Api::get_auth_info`] which
-/// knows what to do when we get [`None`] instead of [`AuthInfo`].
-pub async fn get_auth_info(
-    api: &impl Api,
-    extra: &ConsoleReqExtra<'_>,
-    creds: &ClientCredentials<'_>,
-) -> Result<AuthInfo, errors::GetAuthInfoError> {
-    info!("fetching user's authentication info");
-    let info = api.get_auth_info(extra, creds).await?.unwrap_or_else(|| {
-        // If we don't have an authentication secret, we mock one to
-        // prevent malicious probing (possible due to missing protocol steps).
-        // This mocked secret will never lead to successful authentication.
-        info!("authentication info not found, mocking it");
-        AuthInfo::Scram(scram::ServerSecret::mock(creds.user, rand::random()))
-    });
-
-    Ok(info)
+/// Various caches for [`console`].
+pub struct ApiCaches {
+    /// Cache for the `wake_compute` API method.
+    pub node_info: NodeInfoCache,
 }

proxy/src/proxy.rs

@@ -2,7 +2,7 @@
 mod tests;
 
 use crate::{
-    auth::{self, backend::AuthSuccess, credentials},
+    auth::{self, backend::AuthSuccess},
     cancellation::{self, CancelMap},
     compute::{self, PostgresConnection},
     config::{ProxyConfig, TlsConfig},
@@ -112,6 +112,7 @@ pub async fn handle_ws_client(
     }
 
     let tls = config.tls_config.as_ref();
+    let hostname = hostname.as_deref();
 
     // TLS is None here, because the connection is already encrypted.
     let do_handshake = handshake(stream, None, cancel_map);
@@ -120,17 +121,13 @@ pub async fn handle_ws_client(
         None => return Ok(()), // it's a cancellation request
     };
 
-    let sni = credentials::SniParams {
-        sni: hostname.as_deref(),
-        common_name: tls.and_then(|tls| tls.common_name.as_deref()),
-    };
-
     // Extract credentials which we're going to use for auth.
     let creds = {
+        let common_name = tls.and_then(|tls| tls.common_name.as_deref());
         let result = config
             .auth_backend
             .as_ref()
-            .map(|_| auth::ClientCredentials::parse(&params, &sni))
+            .map(|_| auth::ClientCredentials::parse(&params, hostname, common_name))
             .transpose();
 
         async { result }.or_else(|e| stream.throw_error(e)).await?
@@ -162,17 +159,14 @@ async fn handle_client(
         None => return Ok(()), // it's a cancellation request
     };
 
-    let sni = credentials::SniParams {
-        sni: stream.get_ref().sni_hostname(),
-        common_name: tls.and_then(|tls| tls.common_name.as_deref()),
-    };
-
     // Extract credentials which we're going to use for auth.
     let creds = {
+        let sni = stream.get_ref().sni_hostname();
+        let common_name = tls.and_then(|tls| tls.common_name.as_deref());
         let result = config
             .auth_backend
             .as_ref()
-            .map(|_| auth::ClientCredentials::parse(&params, &sni))
+            .map(|_| auth::ClientCredentials::parse(&params, sni, common_name))
             .transpose();
 
         async { result }.or_else(|e| stream.throw_error(e)).await?

proxy/src/proxy/tests.rs

@@ -92,10 +92,10 @@ impl TestAuth for NoAuth {}
 struct Scram(scram::ServerSecret);
 
 impl Scram {
-    fn new(password: &[u8]) -> anyhow::Result<Self> {
+    fn new(password: &str) -> anyhow::Result<Self> {
         let salt = rand::random::<[u8; 16]>();
-        let secret = scram::ServerSecret::build(password, &salt, 256);
-
+        let secret = scram::ServerSecret::build(password, &salt, 256)
+            .context("failed to generate scram secret")?;
         Ok(Scram(secret))
     }
 
@@ -230,11 +230,11 @@ async fn keepalive_is_inherited() -> anyhow::Result<()> {
 }
 
 #[rstest]
-#[case(b"password_foo")]
-#[case(b"pwd-bar")]
-#[case(b"")]
+#[case("password_foo")]
+#[case("pwd-bar")]
+#[case("")]
 #[tokio::test]
-async fn scram_auth_good(#[case] password: &[u8]) -> anyhow::Result<()> {
+async fn scram_auth_good(#[case] password: &str) -> anyhow::Result<()> {
     let (client, server) = tokio::io::duplex(1024);
 
     let (client_config, server_config) =

proxy/src/scram.rs

@@ -12,6 +12,7 @@ mod messages;
 mod secret;
 mod signature;
 
+#[cfg(test)]
 mod password;
 
 pub use exchange::Exchange;

proxy/src/scram/exchange.rs

@@ -73,7 +73,7 @@ impl sasl::Mechanism for Exchange<'_> {
 
                 let server_first_message = client_first_message.build_server_first_message(
                     &(self.nonce)(),
-                    &self.secret.salt,
+                    &self.secret.salt_base64,
                     self.secret.iterations,
                 );
                 let msg = server_first_message.as_str().to_owned();

proxy/src/scram/messages.rs

@@ -75,27 +75,19 @@ impl<'a> ClientFirstMessage<'a> {
     pub fn build_server_first_message(
         &self,
         nonce: &[u8; SCRAM_RAW_NONCE_LEN],
-        salt: &[u8],
+        salt_base64: &str,
         iterations: u32,
     ) -> OwnedServerFirstMessage {
         use std::fmt::Write;
 
         let mut message = String::new();
-
-        // Write base64-encoded combined nonce.
         write!(&mut message, "r={}", self.nonce).unwrap();
         base64::encode_config_buf(nonce, base64::STANDARD, &mut message);
         let combined_nonce = 2..message.len();
-
-        // Write base64-encoded salt.
-        write!(&mut message, ",s=").unwrap();
-        base64::encode_config_buf(salt, base64::STANDARD, &mut message);
-
-        // Write number of iterations.
-        write!(&mut message, ",i={iterations}").unwrap();
+        write!(&mut message, ",s={},i={}", salt_base64, iterations).unwrap();
 
         // This design guarantees that it's impossible to create a
-        // server-first-message without receiving a client-first-message.
+        // server-first-message without receiving a client-first-message
         OwnedServerFirstMessage {
             message,
             nonce: combined_nonce,
@@ -237,49 +229,4 @@ mod tests {
             "SRpfsIVS4Gk11w1LqQ4QvCUBZYQmqXNSDEcHqbQ3CHI="
         );
     }
-
-    #[test]
-    fn build_server_messages() {
-        let input = "n,,n=pepe,r=t8JwklwKecDLwSsA72rHmVju";
-        let client_first_message = ClientFirstMessage::parse(input).unwrap();
-
-        let nonce = [0; 18];
-        let salt = [1, 2, 3];
-        let iterations = 4096;
-        let server_first_message =
-            client_first_message.build_server_first_message(&nonce, &salt, iterations);
-
-        assert_eq!(
-            server_first_message.message,
-            "r=t8JwklwKecDLwSsA72rHmVjuAAAAAAAAAAAAAAAAAAAAAAAA,s=AQID,i=4096"
-        );
-        assert_eq!(
-            server_first_message.nonce(),
-            "t8JwklwKecDLwSsA72rHmVjuAAAAAAAAAAAAAAAAAAAAAAAA"
-        );
-
-        let input = [
-            "c=eSws",
-            "r=iiYEfS3rOgn8S3rtpSdrOsHtPLWvIkdgmHxA0hf3JNOAG4dU",
-            "p=SRpfsIVS4Gk11w1LqQ4QvCUBZYQmqXNSDEcHqbQ3CHI=",
-        ]
-        .join(",");
-
-        let client_final_message = ClientFinalMessage::parse(&input).unwrap();
-
-        let signature_builder = SignatureBuilder {
-            client_first_message_bare: client_first_message.bare,
-            server_first_message: server_first_message.as_str(),
-            client_final_message_without_proof: client_final_message.without_proof,
-        };
-
-        let server_key = ScramKey::default();
-        let server_final_message =
-            client_final_message.build_server_final_message(signature_builder, &server_key);
-
-        assert_eq!(
-            server_final_message,
-            "v=XEL4X1vy5LnqIgOo4hOjm7zd1Ceyo9+nBUE+/zVnqLE="
-        );
-    }
 }

proxy/src/scram/password.rs

@@ -1,7 +1,6 @@
 //! Password hashing routines.
 
 use super::key::ScramKey;
-use tracing::warn;
 
 pub const SALTED_PASSWORD_LEN: usize = 32;
 
@@ -14,12 +13,7 @@ pub struct SaltedPassword {
 impl SaltedPassword {
     /// See `scram-common.c : scram_SaltedPassword` for details.
     /// Further reading: <https://datatracker.ietf.org/doc/html/rfc2898> (see `PBKDF2`).
-    /// TODO: implement proper password normalization required by the RFC!
     pub fn new(password: &[u8], salt: &[u8], iterations: u32) -> SaltedPassword {
-        if !password.is_ascii() {
-            warn!("found non-ascii symbols in password! salted password might be broken");
-        }
-
         let one = 1_u32.to_be_bytes(); // magic
 
         let mut current = super::hmac_sha256(password, [salt, &one]);
@@ -36,7 +30,6 @@ impl SaltedPassword {
     }
 
     /// Derive `ClientKey` from a salted hashed password.
-    #[cfg(test)]
     pub fn client_key(&self) -> ScramKey {
         super::hmac_sha256(&self.bytes, [b"Client Key".as_ref()]).into()
     }

proxy/src/scram/secret.rs

@@ -1,14 +1,15 @@
 //! Tools for SCRAM server secret management.
 
-use super::{base64_decode_array, key::ScramKey, password::SaltedPassword};
+use super::base64_decode_array;
+use super::key::ScramKey;
 
-/// Server secret is produced from [password](SaltedPassword)
+/// Server secret is produced from [password](super::password::SaltedPassword)
 /// and is used throughout the authentication process.
 pub struct ServerSecret {
     /// Number of iterations for `PBKDF2` function.
     pub iterations: u32,
     /// Salt used to hash user's password.
-    pub salt: Vec<u8>,
+    pub salt_base64: String,
     /// Hashed `ClientKey`.
     pub stored_key: ScramKey,
     /// Used by client to verify server's signature.
@@ -29,7 +30,7 @@ impl ServerSecret {
 
         let secret = ServerSecret {
             iterations: iterations.parse().ok()?,
-            salt: base64::decode(salt).ok()?,
+            salt_base64: salt.to_owned(),
             stored_key: base64_decode_array(stored_key)?.into(),
             server_key: base64_decode_array(server_key)?.into(),
             doomed: false,
@@ -47,31 +48,31 @@ impl ServerSecret {
 
         Self {
             iterations: 4096,
-            salt: mocked_salt.into(),
+            salt_base64: base64::encode(mocked_salt),
             stored_key: ScramKey::default(),
             server_key: ScramKey::default(),
             doomed: true,
         }
     }
 
-    /// Check if this secret was derived from the given password.
-    pub fn matches_password(&self, password: &[u8]) -> bool {
-        let password = SaltedPassword::new(password, &self.salt, self.iterations);
-        self.server_key == password.server_key()
-    }
-
     /// Build a new server secret from the prerequisites.
+    /// XXX: We only use this function in tests.
     #[cfg(test)]
-    pub fn build(password: &[u8], salt: &[u8], iterations: u32) -> Self {
-        let password = SaltedPassword::new(password, salt, iterations);
+    pub fn build(password: &str, salt: &[u8], iterations: u32) -> Option<Self> {
+        // TODO: implement proper password normalization required by the RFC
+        if !password.is_ascii() {
+            return None;
+        }
 
-        Self {
+        let password = super::password::SaltedPassword::new(password.as_bytes(), salt, iterations);
+
+        Some(Self {
             iterations,
-            salt: salt.into(),
+            salt_base64: base64::encode(salt),
             stored_key: password.client_key().sha256(),
             server_key: password.server_key(),
             doomed: false,
-        }
+        })
     }
 }
 
@@ -86,11 +87,17 @@ mod tests {
         let stored_key = "D5h6KTMBlUvDJk2Y8ELfC1Sjtc6k9YHjRyuRZyBNJns=";
         let server_key = "Pi3QHbcluX//NDfVkKlFl88GGzlJ5LkyPwcdlN/QBvI=";
 
-        let secret = format!("SCRAM-SHA-256${iterations}:{salt}${stored_key}:{server_key}");
+        let secret = format!(
+            "SCRAM-SHA-256${iterations}:{salt}${stored_key}:{server_key}",
+            iterations = iterations,
+            salt = salt,
+            stored_key = stored_key,
+            server_key = server_key,
+        );
 
         let parsed = ServerSecret::parse(&secret).unwrap();
         assert_eq!(parsed.iterations, iterations);
-        assert_eq!(base64::encode(parsed.salt), salt);
+        assert_eq!(parsed.salt_base64, salt);
 
         assert_eq!(base64::encode(parsed.stored_key), stored_key);
         assert_eq!(base64::encode(parsed.server_key), server_key);
@@ -99,9 +106,9 @@ mod tests {
     #[test]
     fn build_scram_secret() {
         let salt = b"salt";
-        let secret = ServerSecret::build(b"password", salt, 4096);
+        let secret = ServerSecret::build("password", salt, 4096).unwrap();
         assert_eq!(secret.iterations, 4096);
-        assert_eq!(secret.salt, salt);
+        assert_eq!(secret.salt_base64, base64::encode(salt));
         assert_eq!(
             base64::encode(secret.stored_key.as_ref()),
             "lF4cRm/Jky763CN4HtxdHnjV4Q8AWTNlKvGmEFFU8IQ="
@@ -111,12 +118,4 @@ mod tests {
             "ub8OgRsftnk2ccDMOt7ffHXNcikRkQkq1lh4xaAqrSw="
         );
     }
-
-    #[test]
-    fn secret_match_password() {
-        let password = b"password";
-        let secret = ServerSecret::build(password, b"salt", 2);
-        assert!(secret.matches_password(password));
-        assert!(!secret.matches_password(b"different"));
-    }
 }