security: isolate pgwire connection failures

Author: farhan-syah

nodedb-cluster-tests/tests/transport_security.rs

@@ -38,12 +38,12 @@ fn insecure_counter_guard() -> &'static Mutex<()> {
     LOCK.get_or_init(|| Mutex::new(()))
 }
 
+use nodedb_cluster::topology::{ClusterTopology, NodeInfo, NodeState};
 use nodedb_cluster::transport::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
 use nodedb_cluster::{
     MacKey, NexarTransport, RaftRpcHandler, TlsCredentials, TransportCredentials,
     generate_node_credentials, insecure_transport_count, spki_pin_from_cert_der,
 };
-use nodedb_cluster::topology::{ClusterTopology, NodeInfo, NodeState};
 use nodedb_raft::message::{AppendEntriesRequest, AppendEntriesResponse, RequestVoteResponse};
 use nodedb_raft::transport::RaftTransport;
 

nodedb/src/bootstrap/mod.rs

@@ -7,6 +7,7 @@ pub mod credentials;
 pub mod data_group_recovery;
 pub mod data_plane;
 pub mod listeners;
+pub mod panic_hook;
 pub mod schema_rehydrate;
 pub mod signal;
 pub mod state_wiring;

nodedb/src/bootstrap/panic_hook.rs

@@ -0,0 +1,66 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Process-wide panic reporting that never renders application panic payloads.
+
+use std::sync::OnceLock;
+
+static PANIC_HOOK: OnceLock<()> = OnceLock::new();
+const PANIC_DIAGNOSTIC: &[u8] = b"nodedb: process panic intercepted\n";
+
+#[cfg(unix)]
+fn report_panic() {
+    // SAFETY: the byte slice is valid for the duration of the call and
+    // `STDERR_FILENO` is the conventional process stderr descriptor. `write`
+    // is allocation-free; failures and partial writes are intentionally
+    // ignored because panic reporting must never trigger another panic.
+    let _ = unsafe {
+        libc::write(
+            libc::STDERR_FILENO,
+            PANIC_DIAGNOSTIC.as_ptr().cast(),
+            PANIC_DIAGNOSTIC.len(),
+        )
+    };
+}
+
+#[cfg(not(unix))]
+fn report_panic() {
+    use std::io::Write as _;
+
+    // The portable fallback performs no formatting and ignores every I/O
+    // failure. NodeDB's supported production targets use the Unix path above.
+    let _ = std::io::stderr().write_all(PANIC_DIAGNOSTIC);
+}
+
+/// Install the payload-redacting process panic hook exactly once.
+///
+/// The hook intentionally neither chains the default hook nor examines the
+/// panic payload or source location. Connection-level boundaries handle
+/// expected wire panics; this fixed diagnostic is the last-resort report for
+/// every other task.
+pub fn install() {
+    PANIC_HOOK.get_or_init(|| {
+        std::panic::set_hook(Box::new(|_| report_panic()));
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn local_install_state_is_idempotent_without_replacing_global_hook() {
+        let state = OnceLock::new();
+        assert!(state.set(()).is_ok());
+        assert!(state.set(()).is_err());
+    }
+
+    #[test]
+    fn diagnostic_is_fixed_and_payload_free() {
+        assert_eq!(PANIC_DIAGNOSTIC, b"nodedb: process panic intercepted\n");
+        assert!(
+            !PANIC_DIAGNOSTIC
+                .windows(b"secret panic payload".len())
+                .any(|window| window == b"secret panic payload")
+        );
+    }
+}

nodedb/src/control/notify_bus.rs

@@ -247,9 +247,11 @@ pub fn normalize_channel(channel: &str) -> String {
 
 /// Handle for a session's LISTEN subscriptions.
 ///
-/// Stores `(channel, session_id, receiver)` triples so the session can
-/// drain notifications and clean up on disconnect.
+/// Stores the immutable subscription tenant together with `(channel,
+/// session_id, receiver)` so disconnect cleanup cannot be redirected by a
+/// later session-identity change.
 pub struct ListenHandle {
+    pub tenant_id: TenantId,
     pub channel: String,
     pub session_id: u64,
     pub rx: tokio::sync::mpsc::Receiver<Notification>,

nodedb/src/control/server/native/dispatch/direct_ops.rs

@@ -354,15 +354,21 @@ async fn dispatch_single_task(
     // response the same way the non-staged branch below shapes it.
     let plan_for_staged_response = task.plan.clone();
 
-    let task = match route_in_tx_write(ctx.state, ctx.sessions, ctx.peer_addr, task, |stage_task| {
-        dispatch_authorized_single_task(
-            ctx,
-            stage_task.tenant_id,
-            stage_task.vshard_id,
-            stage_task.plan,
-            stage_task.txn_id,
-        )
-    })
+    let task = match route_in_tx_write(
+        ctx.state,
+        ctx.sessions,
+        ctx.peer_addr.into(),
+        task,
+        |stage_task| {
+            dispatch_authorized_single_task(
+                ctx,
+                stage_task.tenant_id,
+                stage_task.vshard_id,
+                stage_task.plan,
+                stage_task.txn_id,
+            )
+        },
+    )
     .await
     {
         Ok(InTxnRoute::Read(routed_task)) => *routed_task,
@@ -420,7 +426,7 @@ async fn dispatch_single_task(
                 crate::control::server::shared::session::record_reads_for_response(
                     ctx.state,
                     ctx.sessions,
-                    ctx.peer_addr,
+                    ctx.peer_addr.into(),
                     ctx.tenant_id(),
                     crate::control::server::shared::session::ResponseReads {
                         plan: &plan_for_response,

nodedb/src/control/server/native/dispatch/graph_match.rs

@@ -78,7 +78,7 @@ pub(crate) async fn handle_graph_match(
         crate::control::server::shared::session::record_reads_for_response(
             ctx.state,
             ctx.sessions,
-            ctx.peer_addr,
+            ctx.peer_addr.into(),
             ctx.tenant_id(),
             crate::control::server::shared::session::ResponseReads {
                 plan: &plan_for_response,

nodedb/src/control/server/native/dispatch/sql.rs

@@ -146,7 +146,7 @@ async fn handle_sql_inner(
     // DDL: try DDL router first.
     let txn_ctx = crate::control::server::shared::session::DmlTxnCtx {
         sessions: ctx.sessions,
-        addr: ctx.peer_addr,
+        session_id: ctx.peer_addr.into(),
     };
     if let Some(result) = crate::control::server::shared::ddl::dispatch(
         ctx.state,

nodedb/src/control/server/native/dispatch/sql_loop.rs

@@ -87,7 +87,7 @@ pub(super) async fn run_dispatch_loop(
         let routed = match route_in_tx_expander(
             ctx.state,
             ctx.sessions,
-            ctx.peer_addr,
+            ctx.peer_addr.into(),
             task,
             |stage_task| async move {
                 dispatch_task(ctx, stage_task)
@@ -102,7 +102,7 @@ pub(super) async fn run_dispatch_loop(
                 route_in_tx_write(
                     ctx.state,
                     ctx.sessions,
-                    ctx.peer_addr,
+                    ctx.peer_addr.into(),
                     *task,
                     |stage_task| async move {
                         dispatch_task(ctx, stage_task)
@@ -192,7 +192,7 @@ pub(super) async fn run_dispatch_loop(
             crate::control::server::shared::session::record_reads_for_response(
                 ctx.state,
                 ctx.sessions,
-                ctx.peer_addr,
+                ctx.peer_addr.into(),
                 ctx.tenant_id(),
                 crate::control::server::shared::session::ResponseReads {
                     plan: &plan_for_response,

nodedb/src/control/server/native/dispatch/transaction.rs

@@ -74,7 +74,7 @@ pub(crate) fn handle_begin(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
     // ensure the session exists first (SQL "BEGIN" already does this in
     // `sql.rs` before calling here).
     ctx.sessions.ensure_session(*ctx.peer_addr);
-    match lifecycle::run_begin(ctx.sessions, ctx.peer_addr, ctx.state) {
+    match lifecycle::run_begin(ctx.sessions, ctx.peer_addr.into(), ctx.state) {
         Ok(()) => NativeResponse::status_row(seq, "BEGIN"),
         Err(e) => {
             let message = match &e {
@@ -88,15 +88,30 @@ pub(crate) fn handle_begin(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
 
 pub(crate) async fn handle_commit(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
     let dp = NativeTxnDp { state: ctx.state };
-    match commit::run_commit(ctx.sessions, ctx.peer_addr, ctx.identity, ctx.state, &dp).await {
+    match commit::run_commit(
+        ctx.sessions,
+        ctx.peer_addr.into(),
+        ctx.identity,
+        ctx.state,
+        &dp,
+    )
+    .await
+    {
         CommitOutcome::Committed => NativeResponse::status_row(seq, "COMMIT"),
         CommitOutcome::Aborted { reason } => commit_abort_to_native(seq, &reason),
     }
 }
 
 pub(crate) async fn handle_rollback(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
     let dp = NativeTxnDp { state: ctx.state };
-    lifecycle::run_rollback(ctx.sessions, ctx.peer_addr, ctx.identity, ctx.state, &dp).await;
+    lifecycle::run_rollback(
+        ctx.sessions,
+        ctx.peer_addr.into(),
+        ctx.identity,
+        ctx.state,
+        &dp,
+    )
+    .await;
     NativeResponse::status_row(seq, "ROLLBACK")
 }
 

nodedb/src/control/server/native/dispatch/transaction_savepoint.rs

@@ -38,8 +38,14 @@ pub(crate) async fn handle_savepoint(
 ) -> NativeResponse {
     let sp_name = sql_trimmed.split_whitespace().nth(1).unwrap_or("sp");
     let dp = NativeTxnDp { state: ctx.state };
-    match savepoint_ops::run_savepoint(ctx.sessions, ctx.peer_addr, ctx.tenant_id(), &dp, sp_name)
-        .await
+    match savepoint_ops::run_savepoint(
+        ctx.sessions,
+        ctx.peer_addr.into(),
+        ctx.tenant_id(),
+        &dp,
+        sp_name,
+    )
+    .await
     {
         Ok(()) => NativeResponse::status_row(seq, "SAVEPOINT"),
         Err(e) => savepoint_error_to_native(seq, &e),
@@ -53,7 +59,7 @@ pub(crate) fn handle_release_savepoint(
     sql_trimmed: &str,
 ) -> NativeResponse {
     let sp_name = sql_trimmed.split_whitespace().last().unwrap_or("sp");
-    match savepoint_ops::run_release_savepoint(ctx.sessions, ctx.peer_addr, sp_name) {
+    match savepoint_ops::run_release_savepoint(ctx.sessions, ctx.peer_addr.into(), sp_name) {
         Ok(()) => NativeResponse::status_row(seq, "RELEASE"),
         Err(e) => savepoint_error_to_native(seq, &e),
     }
@@ -69,7 +75,7 @@ pub(crate) async fn handle_rollback_to_savepoint(
     let dp = NativeTxnDp { state: ctx.state };
     match savepoint_ops::run_rollback_to_savepoint(
         ctx.sessions,
-        ctx.peer_addr,
+        ctx.peer_addr.into(),
         ctx.tenant_id(),
         &dp,
         sp_name,