Skip to content

Commit 6256bec

Browse files
committed
Merge branch 'security': isolate pgwire and non-pgwire connection failures
2 parents 7d0dfac + 22e70d2 commit 6256bec

114 files changed

Lines changed: 4200 additions & 1540 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

nodedb/src/bootstrap/mod.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ pub mod credentials;
77
pub mod data_group_recovery;
88
pub mod data_plane;
99
pub mod listeners;
10+
pub mod panic_hook;
1011
pub mod schema_rehydrate;
1112
pub mod signal;
1213
pub mod state_wiring;

nodedb/src/bootstrap/panic_hook.rs

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
// SPDX-License-Identifier: BUSL-1.1
2+
3+
//! Process-wide panic reporting that never renders application panic payloads.
4+
5+
use std::sync::OnceLock;
6+
7+
static PANIC_HOOK: OnceLock<()> = OnceLock::new();
8+
const PANIC_DIAGNOSTIC: &[u8] = b"nodedb: process panic intercepted\n";
9+
10+
#[cfg(unix)]
11+
fn report_panic() {
12+
// SAFETY: the byte slice is valid for the duration of the call and
13+
// `STDERR_FILENO` is the conventional process stderr descriptor. `write`
14+
// is allocation-free; failures and partial writes are intentionally
15+
// ignored because panic reporting must never trigger another panic.
16+
let _ = unsafe {
17+
libc::write(
18+
libc::STDERR_FILENO,
19+
PANIC_DIAGNOSTIC.as_ptr().cast(),
20+
PANIC_DIAGNOSTIC.len(),
21+
)
22+
};
23+
}
24+
25+
#[cfg(not(unix))]
26+
fn report_panic() {
27+
use std::io::Write as _;
28+
29+
// The portable fallback performs no formatting and ignores every I/O
30+
// failure. NodeDB's supported production targets use the Unix path above.
31+
let _ = std::io::stderr().write_all(PANIC_DIAGNOSTIC);
32+
}
33+
34+
/// Install the payload-redacting process panic hook exactly once.
35+
///
36+
/// The hook intentionally neither chains the default hook nor examines the
37+
/// panic payload or source location. Connection-level boundaries handle
38+
/// expected wire panics; this fixed diagnostic is the last-resort report for
39+
/// every other task.
40+
pub fn install() {
41+
PANIC_HOOK.get_or_init(|| {
42+
std::panic::set_hook(Box::new(|_| report_panic()));
43+
});
44+
}
45+
46+
#[cfg(test)]
47+
mod tests {
48+
use super::*;
49+
50+
#[test]
51+
fn local_install_state_is_idempotent_without_replacing_global_hook() {
52+
let state = OnceLock::new();
53+
assert!(state.set(()).is_ok());
54+
assert!(state.set(()).is_err());
55+
}
56+
57+
#[test]
58+
fn diagnostic_is_fixed_and_payload_free() {
59+
assert_eq!(PANIC_DIAGNOSTIC, b"nodedb: process panic intercepted\n");
60+
assert!(
61+
!PANIC_DIAGNOSTIC
62+
.windows(b"secret panic payload".len())
63+
.any(|window| window == b"secret panic payload")
64+
);
65+
}
66+
}

nodedb/src/control/notify_bus.rs

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -247,9 +247,11 @@ pub fn normalize_channel(channel: &str) -> String {
247247

248248
/// Handle for a session's LISTEN subscriptions.
249249
///
250-
/// Stores `(channel, session_id, receiver)` triples so the session can
251-
/// drain notifications and clean up on disconnect.
250+
/// Stores the immutable subscription tenant together with `(channel,
251+
/// session_id, receiver)` so disconnect cleanup cannot be redirected by a
252+
/// later session-identity change.
252253
pub struct ListenHandle {
254+
pub tenant_id: TenantId,
253255
pub channel: String,
254256
pub session_id: u64,
255257
pub rx: tokio::sync::mpsc::Receiver<Notification>,

nodedb/src/control/server/http/routes/crdt.rs

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ pub async fn crdt_apply(
115115
// replication. A local-only dispatch would land it on the receiving node only
116116
// — lost to followers and entirely on leader failover. This handler is scoped
117117
// to the default database (matching its surrogate assignment above).
118-
state.shared.tenant_request_start(identity.tenant_id);
118+
let _request = state.shared.tenant_request_guard(identity.tenant_id);
119119
let policy = crate::control::crdt_post_image_policy::ExternalCrdtPostImagePolicy::from_identity(
120120
identity.tenant_id,
121121
crate::types::DatabaseId::DEFAULT,
@@ -138,7 +138,6 @@ pub async fn crdt_apply(
138138
},
139139
)
140140
.await;
141-
state.shared.tenant_request_end(identity.tenant_id);
142141

143142
result.map_err(ApiError::from)?;
144143

nodedb/src/control/server/http/routes/query/materialized.rs

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -129,12 +129,12 @@ pub async fn query(
129129
}
130130

131131
// Track active request for quota accounting.
132-
state.shared.tenant_request_start(tenant_id);
132+
let _request = state.shared.tenant_request_guard(tenant_id);
133133

134134
// Execute each task via the SPSC bridge.
135135
let mut result_rows = Vec::new();
136136

137-
let result = async {
137+
async {
138138
for (task, authorized_task) in tasks.into_iter().zip(authorized_tasks) {
139139
// `INSERT ... SELECT` is orchestrated on the Control Plane: the
140140
// source is scanned, each target row gets its OWN fresh, registered
@@ -308,10 +308,7 @@ pub async fn query(
308308

309309
Ok(axum::Json(HttpQueryResponse::ok(result_rows)))
310310
}
311-
.await;
312-
313-
state.shared.tenant_request_end(tenant_id);
314-
result
311+
.await
315312
}
316313

317314
fn ddl_error_to_api(error: crate::control::server::shared::ddl::DdlError) -> ApiError {

nodedb/src/control/server/http/routes/query/ndjson.rs

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -152,7 +152,7 @@ pub async fn query_ndjson(
152152
Err(error) => return ApiError::from(crate::Error::from(error)).into_response(),
153153
};
154154

155-
state.shared.tenant_request_start(tenant_id);
155+
let _request = state.shared.tenant_request_guard(tenant_id);
156156

157157
let mut ndjson = String::new();
158158
for (task, authorized_task) in tasks.into_iter().zip(authorized_tasks) {
@@ -262,8 +262,6 @@ pub async fn query_ndjson(
262262
}
263263
}
264264

265-
state.shared.tenant_request_end(tenant_id);
266-
267265
Response::builder()
268266
.header("Content-Type", "application/x-ndjson")
269267
.body(axum::body::Body::from(ndjson))

nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs

Lines changed: 5 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ pub async fn execute_sql(
7070
)?
7171
.into_tasks();
7272

73-
shared.tenant_request_start(tenant_id);
73+
let _request = shared.tenant_request_guard(tenant_id);
7474

7575
let mut results = Vec::new();
7676
for (task, authorized_task) in tasks.into_iter().zip(authorized_tasks) {
@@ -98,10 +98,7 @@ pub async fn execute_sql(
9898
}
9999
}
100100
}
101-
Err(e) => {
102-
shared.tenant_request_end(tenant_id);
103-
return Err(e);
104-
}
101+
Err(e) => return Err(e),
105102
}
106103
continue;
107104
}
@@ -138,10 +135,7 @@ pub async fn execute_sql(
138135
}
139136
}
140137
}
141-
Err(e) => {
142-
shared.tenant_request_end(tenant_id);
143-
return Err(e);
144-
}
138+
Err(e) => return Err(e),
145139
}
146140
continue;
147141
}
@@ -182,10 +176,7 @@ pub async fn execute_sql(
182176
}
183177
}
184178
}
185-
Err(e) => {
186-
shared.tenant_request_end(tenant_id);
187-
return Err(e);
188-
}
179+
Err(e) => return Err(e),
189180
}
190181
continue;
191182
}
@@ -225,15 +216,10 @@ pub async fn execute_sql(
225216
}
226217
}
227218
}
228-
Err(e) => {
229-
shared.tenant_request_end(tenant_id);
230-
return Err(e);
231-
}
219+
Err(e) => return Err(e),
232220
}
233221
}
234222

235-
shared.tenant_request_end(tenant_id);
236-
237223
match results.len() {
238224
0 => Ok(serde_json::Value::Null),
239225
1 => Ok(results

0 commit comments

Comments
 (0)