diff --git a/Cargo.lock b/Cargo.lock index d1a7e2b0b..9970838bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -514,6 +514,7 @@ dependencies = [ "corepc-types", "crossbeam-channel", "hashbrown 0.17.1", + "metrics", "miniscript", "parking_lot", "rustix", diff --git a/bin/bitcoin-rs/src/main.rs b/bin/bitcoin-rs/src/main.rs index c99df5a09..09d2c86af 100644 --- a/bin/bitcoin-rs/src/main.rs +++ b/bin/bitcoin-rs/src/main.rs @@ -288,6 +288,39 @@ mod tests { ); } + #[test] + fn public_listener_cli_overrides_environment_and_file() -> anyhow::Result<()> { + let config = load_file( + "--config", + "public_bind = '127.0.0.1:8080'\nrpc_max_connections = 4\npublic_max_connections = 32\n", + &[ + "--public-bind", + "0.0.0.0:9090", + "--public-max-connections", + "16", + ], + [ + ( + OsString::from("BITCOIN_RS_PUBLIC_BIND"), + OsString::from("127.0.0.1:8081"), + ), + ( + OsString::from("BITCOIN_RS_RPC_MAX_CONNECTIONS"), + OsString::from("8"), + ), + ( + OsString::from("BITCOIN_RS_PUBLIC_MAX_CONNECTIONS"), + OsString::from("24"), + ), + ] + .into_iter(), + )?; + assert_eq!(config.rpc.public_bind, Some("0.0.0.0:9090".parse()?)); + assert_eq!(config.rpc.max_connections, 8); + assert_eq!(config.rpc.public_max_connections, 16); + Ok(()) + } + #[test] fn environment_parses_script_index() { let config = resolved(&["bitcoin-rs"], &[("BITCOIN_RS_SCRIPTINDEX", "full")]); diff --git a/crates/node/src/config.rs b/crates/node/src/config.rs index 9f16d4886..5837c7b18 100644 --- a/crates/node/src/config.rs +++ b/crates/node/src/config.rs @@ -159,6 +159,12 @@ pub struct P2pConfig { #[derive(Clone, Debug)] pub struct RpcConfig { pub bind: SocketAddr, + /// Opt-in separate public REST/Esplora listener. + pub public_bind: Option, + /// Connection budget of the shared or management listener. + pub max_connections: usize, + /// Independent connection budget of the public listener. + pub public_max_connections: usize, pub rest: bool, /// Optional experimental keyless BIP352 read service, off by default. pub silent_payments: bool, @@ -288,6 +294,9 @@ impl NodeConfig { }, rpc: RpcConfig { bind: settings.rpc.bind.unwrap_or(profile.rpc_bind), + public_bind: settings.rpc.public_bind, + max_connections: settings.rpc.max_connections.unwrap_or(128), + public_max_connections: settings.rpc.public_max_connections.unwrap_or(128), rest: settings.rpc.rest.unwrap_or(false), silent_payments: settings.rpc.silent_payments.unwrap_or(false), auth: rpc_auth(settings), @@ -366,6 +375,26 @@ impl NodeConfig { self.rpc.bind.ip().is_loopback() || self.rpc.auth != Auth::GeneratedCookie, "non-loopback RPC bind requires explicit credentials (--rpc-password or --rpc-cookie)" ); + anyhow::ensure!( + self.rpc.max_connections > 0 && self.rpc.public_max_connections > 0, + "RPC and public connection limits must be positive" + ); + anyhow::ensure!( + self.rpc.public_bind.is_none_or(|address| { + let management = self.rpc.bind; + // Port zero asks the OS for an unused port. Distinct + // concrete IPs may share a port; a same-family wildcard + // overlaps every concrete address on that port. + // Cross-family dual-stack conflicts remain OS-checked at bind. + address.port() == 0 + || address.port() != management.port() + || (address.ip() != management.ip() + && (address.is_ipv4() != management.is_ipv4() + || (!address.ip().is_unspecified() + && !management.ip().is_unspecified()))) + }), + "public-bind and rpc-bind must not overlap" + ); anyhow::ensure!( self.storage.backend.is_compiled_in(), "unsupported storage backend {}", diff --git a/crates/node/src/lifecycle.rs b/crates/node/src/lifecycle.rs index 1a6803536..6612adda4 100644 --- a/crates/node/src/lifecycle.rs +++ b/crates/node/src/lifecycle.rs @@ -7,7 +7,7 @@ use core::num::NonZeroU32; use std::sync::Arc; use std::time::Duration; -use anyhow::Result; +use anyhow::{Context as _, Result}; use bitcoin_rs_chain::{BlockBodySource, LatchReader}; use bitcoin_rs_chainstate::verify::{ @@ -15,7 +15,7 @@ use bitcoin_rs_chainstate::verify::{ }; use bitcoin_rs_mining::MiningControl; use bitcoin_rs_rpc::{ - RpcServer, + RpcServer, RpcSurface, context::{ ChainControl, ChainControlError, ChainHandles, Context, ContextHandles, IndexHandles, MempoolHandles, MiningHandles, NetworkHandles, @@ -59,7 +59,6 @@ mod shutdown { /// poisons the index namespace. pub(crate) const DRAIN_DEADLINE: Duration = Duration::from_secs(5); -const RPC_MAX_CONNECTIONS: usize = 128; const RPC_IDLE_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Clone)] @@ -159,13 +158,18 @@ fn verify_request(check_level: u8, nblocks: u32) -> VerifyRequest { } } -/// Binds the RPC listener without spawning a worker. Startup records the +struct RpcBindings { + context: Arc, + listeners: Vec<(RpcServer, RpcSurface)>, +} + +/// Binds every RPC/HTTP listener before spawning a worker. Startup records each /// worker immediately after spawning, so a later failure cannot detach it. fn bind_rpc( state: &NodeState, mining_control: &Arc, block_body_source: Arc, -) -> Result<(Arc, RpcServer)> { +) -> Result { let config = state.config(); let rpc_auth = Arc::new(config.rpc.auth.to_rpc_auth(&config.data_dir)?); let chainstate = state.chainstate(); @@ -216,13 +220,36 @@ fn bind_rpc( let handler = Arc::new(bitcoin_rs_rpc::Handler::new(Arc::clone(&context))); let server = RpcServer::bind( state.config().rpc.bind, - rpc_auth, - handler, - RPC_MAX_CONNECTIONS, + Arc::clone(&rpc_auth), + Arc::clone(&handler), + config.rpc.max_connections, RPC_IDLE_TIMEOUT, state.config().rpc.rest, - )?; - Ok((context, server)) + ) + .with_context(|| format!("failed to bind RPC listener at {}", config.rpc.bind))?; + let mut servers = Vec::with_capacity(2); + if let Some(address) = config.rpc.public_bind { + // RpcServer's existing binding API requires Auth. Public mode shares + // this object only for API compatibility: its closed route policy + // rejects JSON-RPC before consulting credentials, even valid ones. + let public = RpcServer::bind( + address, + rpc_auth, + handler, + config.rpc.public_max_connections, + RPC_IDLE_TIMEOUT, + config.rpc.rest, + ) + .with_context(|| format!("failed to bind public HTTP listener at {address}"))?; + servers.push((server, RpcSurface::Management)); + servers.push((public, RpcSurface::Public)); + } else { + servers.push((server, RpcSurface::Shared)); + } + Ok(RpcBindings { + context, + listeners: servers, + }) } // Records that teardown reached the bootstrap join on the current thread. @@ -287,7 +314,7 @@ pub(crate) struct NodeServices { event_loop_signal: Option>, /// Stops and joins its listener in Drop. metrics: Option, - rpc_thread: Option>>, + rpc_threads: Vec>>, /// Chainstate idle maintenance (journal flush, retention); joined /// before the final clean checkpoint publication. maintenance_worker: Option>, @@ -367,7 +394,7 @@ impl NodeServices { set_first_error(first_error, anyhow::anyhow!("event loop thread panicked")); } } - if let Some(handle) = self.rpc_thread.take() { + for handle in self.rpc_threads.drain(..) { match handle.join() { Ok(Ok(())) => tracing::info!("rpc listener exited cleanly"), // RPC listener returned an I/O error. @@ -773,21 +800,22 @@ pub(crate) fn start_node( // state. crate::chain_effects::bootstrap_genesis(&chainstate, &state.chain_followers()) .map_err(|error| anyhow::Error::new(error).context("failed to bootstrap genesis"))?; - let (context, rpc_server) = bind_rpc(state, &mining_control, block_body_source)?; - let rpc_local_addr = rpc_server.local_addr()?; - tracing::info!(addr = %rpc_local_addr, "rpc listener bound"); - let rpc_shutdown = state.shutdown_reader(); - guard.services.rpc_thread = Some( - std::thread::Builder::new() + let RpcBindings { context, listeners } = bind_rpc(state, &mining_control, block_body_source)?; + for (rpc_server, surface) in listeners { + let rpc_local_addr = rpc_server.local_addr()?; + tracing::info!(addr = %rpc_local_addr, ?surface, "rpc listener bound"); + let rpc_shutdown = state.shutdown_reader(); + let worker = std::thread::Builder::new() .name("bitcoin-rs-rpc".into()) .spawn(move || { - let result = rpc_server.serve_with_shutdown(rpc_shutdown.clone()); + let result = rpc_server.serve_surface_with_shutdown(rpc_shutdown.clone(), surface); if !rpc_shutdown.is_triggered() { - tracing::error!(error = ?result.as_ref().err(), "rpc listener exited before shutdown"); + tracing::error!(?surface, error = ?result.as_ref().err(), "rpc listener exited before shutdown"); } result - })?, - ); + })?; + guard.services.rpc_threads.push(worker); + } let peer_ready: Arc = Arc::new(move |source: bitcoin_rs_p2p::PeerSource| { peer_ready_sync.on_peer_ready(source); diff --git a/crates/node/src/metrics.rs b/crates/node/src/metrics.rs index dbdc83f63..1cfcce605 100644 --- a/crates/node/src/metrics.rs +++ b/crates/node/src/metrics.rs @@ -230,6 +230,18 @@ fn describe_node_metrics() { TXINDEX_READINESS_GAUGE, "txindex capability readiness from the live getcapabilities source; 1 for the active state label, 0 for the others" ); + metrics::describe_gauge!( + "rpc.connections.active", + "admitted live HTTP connections, by listener surface" + ); + metrics::describe_counter!( + "rpc.connections.rejected_total", + "HTTP connection budget refusals, by listener surface" + ); + metrics::describe_histogram!( + "rpc.http.request_seconds", + "HTTP time from complete request head through response, by listener surface and route family" + ); } static PROMETHEUS_HANDLE: Mutex> = Mutex::new(None); diff --git a/crates/node/src/options.rs b/crates/node/src/options.rs index e0a0664d5..24ffdc1e6 100644 --- a/crates/node/src/options.rs +++ b/crates/node/src/options.rs @@ -587,6 +587,24 @@ macro_rules! option_rows { env["BITCOIN_RS_RPC_BIND", str::parse] toml native("rpc_bind") } + /// Separate public HTTP bind; when set, rpc-bind serves JSON-RPC only. + public_bind as public_bind: Option { + cli[#[arg(long = "public-bind")]] + env["BITCOIN_RS_PUBLIC_BIND", str::parse] + toml native("public_bind") + } + /// Maximum live connections on the RPC listener (default 128). + max_connections as rpc_max_connections: Option { + cli[#[arg(long = "rpc-max-connections")]] + env["BITCOIN_RS_RPC_MAX_CONNECTIONS", str::parse] + toml native("rpc_max_connections") + } + /// Maximum live connections on the optional public listener (default 128). + public_max_connections as public_max_connections: Option { + cli[#[arg(long = "public-max-connections")]] + env["BITCOIN_RS_PUBLIC_MAX_CONNECTIONS", str::parse] + toml native("public_max_connections") + } /// Whether the REST gateway is enabled. rest as rest: Option { cli[#[arg(long = "rest")]] diff --git a/crates/node/tests/config_layered.rs b/crates/node/tests/config_layered.rs index 485f0bd92..91eec8c90 100644 --- a/crates/node/tests/config_layered.rs +++ b/crates/node/tests/config_layered.rs @@ -10,6 +10,71 @@ use bitcoin_rs_node::{ use bitcoin_rs_primitives::Network; use bitcoin_rs_rpc::zmq::ZmqTopic; +#[test] +fn public_listener_is_opt_in_and_budgets_resolve_independently() -> Result<()> { + let defaults = NodeConfig::default_for_network(Network::Regtest); + assert_eq!(defaults.rpc.public_bind, None); + assert_eq!(defaults.rpc.max_connections, 128); + assert_eq!(defaults.rpc.public_max_connections, 128); + let file: UserConfig = toml::from_str( + "public_bind = '0.0.0.0:8080'\nrpc_max_connections = 8\npublic_max_connections = 32", + )?; + let mut environment = UserConfig::default(); + environment.apply_env( + "BITCOIN_RS_PUBLIC_BIND", + std::ffi::OsStr::new("127.0.0.1:9090"), + )?; + environment.apply_env( + "BITCOIN_RS_PUBLIC_MAX_CONNECTIONS", + std::ffi::OsStr::new("16"), + )?; + let config = resolve(&[&file, &environment])?; + assert_eq!(config.rpc.public_bind, Some("127.0.0.1:9090".parse()?)); + assert_eq!(config.rpc.max_connections, 8); + assert_eq!(config.rpc.public_max_connections, 16); + for invalid in [ + "rpc_max_connections = 0", + "public_max_connections = 0", + "rpc_bind = '127.0.0.1:8332'\npublic_bind = '127.0.0.1:8332'", + ] { + let layer: UserConfig = toml::from_str(invalid)?; + assert!(resolve(&[&layer]).is_err(), "{invalid}"); + } + Ok(()) +} + +#[test] +fn public_listener_bind_validation_handles_wildcards_and_ephemeral_ports() -> Result<()> { + for (management, public, valid) in [ + ("127.0.0.1:8332", "127.0.0.1:8332", false), + ("0.0.0.0:8332", "127.0.0.1:8332", false), + ("127.0.0.1:8332", "0.0.0.0:8332", false), + ("[::]:8332", "[::1]:8332", false), + ("[::1]:8332", "[::]:8332", false), + ("127.0.0.1:8332", "127.0.0.2:8332", true), + ("127.0.0.1:8332", "127.0.0.1:8080", true), + ("0.0.0.0:0", "0.0.0.0:0", true), + ("127.0.0.1:0", "127.0.0.1:8332", true), + ("127.0.0.1:8332", "127.0.0.1:0", true), + ] { + let mut config = NodeConfig::default_for_network(Network::Regtest); + config.rpc.bind = management.parse()?; + config.rpc.public_bind = Some(public.parse()?); + config.rpc.auth = bitcoin_rs_node::Auth::basic("operator", "secret"); + let result = config.validate(); + assert_eq!(result.is_ok(), valid, "{management}, {public}: {result:?}"); + if let Err(error) = result { + assert!(error.to_string().contains("must not overlap"), "{error}"); + } + } + // Both sockets stay bound: the second port-zero bind must select a + // different available port rather than reuse the first listener's port. + let management = std::net::TcpListener::bind("0.0.0.0:0")?; + let public = std::net::TcpListener::bind("0.0.0.0:0")?; + assert_ne!(management.local_addr()?.port(), public.local_addr()?.port()); + Ok(()) +} + #[test] fn silent_payments_toml_overlay_preserves_explicit_disable() -> Result<()> { assert!( diff --git a/crates/node/tests/unit/lifecycle/tests.rs b/crates/node/tests/unit/lifecycle/tests.rs index 456f25357..d22308b86 100644 --- a/crates/node/tests/unit/lifecycle/tests.rs +++ b/crates/node/tests/unit/lifecycle/tests.rs @@ -27,6 +27,82 @@ fn isolated_config(data_dir: &Path) -> NodeConfig { config } +#[test] +fn split_rpc_listeners_start_and_join_through_node_services() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let management = std::net::TcpListener::bind("127.0.0.1:0")?; + let public = std::net::TcpListener::bind("127.0.0.1:0")?; + let management_address = management.local_addr()?; + let public_address = public.local_addr()?; + let mut config = isolated_config(dir.path()); + config.rpc.bind = management_address; + config.rpc.public_bind = Some(public_address); + drop((management, public)); + let node = start_node(config, RuntimeInputs::default(), false)?; + assert_eq!( + node.services + .as_ref() + .ok_or_else(|| anyhow::anyhow!("missing node services"))? + .rpc_threads + .len(), + 2 + ); + for (address, request, expected) in [ + ( + public_address, + "OPTIONS /api/tx HTTP/1.1\r\nConnection: close\r\n\r\n", + "HTTP/1.1 204 No Content", + ), + ( + management_address, + "POST / HTTP/1.1\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 401 Unauthorized", + ), + ] { + let mut stream = TcpStream::connect(address)?; + stream.set_read_timeout(Some(WITNESS_BOUND))?; + stream.write_all(request.as_bytes())?; + let mut response = String::new(); + stream.read_to_string(&mut response)?; + assert!(response.starts_with(expected), "{response}"); + } + node.shutdown_blocking()?; + let _management = std::net::TcpListener::bind(management_address)?; + let _public = std::net::TcpListener::bind(public_address)?; + Ok(()) +} + +#[test] +fn public_bind_failure_rolls_back_management_binding_and_node_state() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let management = std::net::TcpListener::bind("127.0.0.1:0")?; + let public = std::net::TcpListener::bind("127.0.0.1:0")?; + let address = management.local_addr()?; + let mut config = isolated_config(dir.path()); + config.rpc.bind = address; + config.rpc.public_bind = Some(public.local_addr()?); + drop(management); + let error = start_node(config.clone(), RuntimeInputs::default(), false) + .err() + .ok_or_else(|| anyhow::anyhow!("occupied public bind must fail startup"))?; + assert_eq!( + error.to_string(), + format!( + "failed to bind public HTTP listener at {}", + public.local_addr()? + ) + ); + assert_eq!( + error + .downcast_ref::() + .map(std::io::Error::kind), + Some(std::io::ErrorKind::AddrInUse) + ); + let _management = std::net::TcpListener::bind(address)?; + let _reopened = NodeState::open(config, None)?; + Ok(()) +} + #[test] fn timed_out_snapshot_worker_cannot_overwrite_the_next_runs_snapshot() -> anyhow::Result<()> { use bitcoin_rs_mempool::{Mempool, MempoolGateway, MempoolLimits, snapshot}; @@ -975,8 +1051,9 @@ fn bind_test_rpc(state: &NodeState) -> anyhow::Result { state.config().mining.payout_script.clone(), )); let mining_control: Arc = coordinator; - let (_context, server) = bind_rpc(state, &mining_control, state.block_body_source()?)?; - Ok(server) + let mut bindings = bind_rpc(state, &mining_control, state.block_body_source()?)?; + assert_eq!(bindings.listeners.len(), 1); + Ok(bindings.listeners.remove(0).0) } #[test] diff --git a/crates/rpc/Cargo.toml b/crates/rpc/Cargo.toml index cfb2cb6c1..af2809e5f 100644 --- a/crates/rpc/Cargo.toml +++ b/crates/rpc/Cargo.toml @@ -40,6 +40,7 @@ compact_str.workspace = true hashbrown.workspace = true thiserror.workspace = true tracing.workspace = true +metrics.workspace = true sha2.workspace = true corepc-types.workspace = true anyhow.workspace = true diff --git a/crates/rpc/src/lib.rs b/crates/rpc/src/lib.rs index d837a34b0..7041309d6 100644 --- a/crates/rpc/src/lib.rs +++ b/crates/rpc/src/lib.rs @@ -44,4 +44,4 @@ pub use manifest::MANIFEST_TOML; pub use error::RpcError; pub use handlers::Handler; -pub use server::RpcServer; +pub use server::{RpcServer, RpcSurface}; diff --git a/crates/rpc/src/server.rs b/crates/rpc/src/server.rs index ac0376e50..5a24c3b56 100644 --- a/crates/rpc/src/server.rs +++ b/crates/rpc/src/server.rs @@ -44,6 +44,36 @@ const PUBLIC_ESPLORA_HEADERS: &[&str] = &["Content-Type"]; const PUBLIC_ESPLORA_EXPOSE_HEADERS: &[&str] = &["X-Total-Results"]; +/// Routes admitted by one listener, with an independent connection budget. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum RpcSurface { + /// Historical combined JSON-RPC, REST and Esplora listener. + #[default] + Shared, + /// Authenticated JSON-RPC only. + Management, + /// REST (when enabled) and both existing Esplora namespaces; no JSON-RPC. + Public, +} + +impl RpcSurface { + const fn label(self) -> &'static str { + match self { + Self::Shared => "shared", + Self::Management => "management", + Self::Public => "public", + } + } + + fn allows(self, route: &HttpRoute<'_>) -> bool { + match self { + Self::Shared => true, + Self::Management => matches!(route, HttpRoute::JsonRpc | HttpRoute::NotFound), + Self::Public => !matches!(route, HttpRoute::JsonRpc), + } + } +} + /// Synchronous HTTP/1.1 JSON-RPC server. pub struct RpcServer { /// Bound TCP listener. @@ -103,12 +133,23 @@ impl RpcServer { self, shutdown: impl Into, ) -> io::Result<()> { - self.serve_accept_loop(&shutdown.into(), TcpListener::accept) + self.serve_surface_with_shutdown(shutdown, RpcSurface::Shared) + } + + /// Runs this listener with a closed route selection and its own budget. + /// The historical serving entrypoint retains the shared surface. + pub fn serve_surface_with_shutdown( + self, + shutdown: impl Into, + surface: RpcSurface, + ) -> io::Result<()> { + self.serve_accept_loop(&shutdown.into(), surface, TcpListener::accept) } fn serve_accept_loop( self, shutdown: &bitcoin_rs_chain::LatchReader, + surface: RpcSurface, mut accept: impl FnMut(&TcpListener) -> io::Result<(TcpStream, SocketAddr)>, ) -> io::Result<()> { let _stop = RpcStop(Arc::clone(&self.handler)); @@ -121,7 +162,7 @@ impl RpcServer { Ok((stream, addr)) => { if let Err(error) = stream .set_nonblocking(false) - .and_then(|()| self.handle_accept(&active, stream, addr.ip())) + .and_then(|()| self.handle_accept(&active, stream, addr.ip(), surface)) { debug!(%error, "rpc connection setup failed"); } @@ -157,6 +198,7 @@ impl RpcServer { active: &Arc>, stream: TcpStream, peer: IpAddr, + surface: RpcSurface, ) -> io::Result<()> { // Connection-local: a nodelay failure must not stop the accept loop. let mut stream = match prepare_http_socket(stream) { @@ -168,6 +210,9 @@ impl RpcServer { }; let should_accept = active.lock().admit(peer, self.max_connections); if !should_accept { + metrics::counter!("rpc.connections.rejected_total", "surface" => surface.label()) + .increment(1); + stream.set_write_timeout(Some(self.idle_timeout))?; write_status(&mut stream, 503, "Service Unavailable", b"busy", false)?; return Ok(()); } @@ -175,7 +220,9 @@ impl RpcServer { let permit = ConnectionPermit { active: Arc::clone(active), peer, + surface, }; + metrics::gauge!("rpc.connections.active", "surface" => surface.label()).increment(1.0); let auth = Arc::clone(&self.auth); let handler = Arc::clone(&self.handler); let rest_enabled = self.rest_enabled; @@ -185,7 +232,7 @@ impl RpcServer { .spawn(move || { let _permit = permit; if let Err(error) = - serve_connection(stream, &auth, &handler, rest_enabled, idle_timeout) + serve_connection(stream, &auth, &handler, rest_enabled, idle_timeout, surface) { debug!(%error, "rpc connection closed with error"); } @@ -240,16 +287,35 @@ impl ActiveConnections { struct ConnectionPermit { active: Arc>, peer: IpAddr, + surface: RpcSurface, } impl Drop for ConnectionPermit { fn drop(&mut self) { self.active.lock().release(self.peer); + metrics::gauge!("rpc.connections.active", "surface" => self.surface.label()).decrement(1.0); } } struct RpcStop(Arc); +struct RequestTimer { + started: Instant, + surface: RpcSurface, + route: &'static str, +} + +impl Drop for RequestTimer { + fn drop(&mut self) { + metrics::histogram!( + "rpc.http.request_seconds", + "surface" => self.surface.label(), + "route" => self.route, + ) + .record(self.started.elapsed().as_secs_f64()); + } +} + impl Drop for RpcStop { fn drop(&mut self) { self.0.stop(); @@ -313,6 +379,7 @@ fn serve_connection( handler: &Handler, rest_enabled: bool, idle_timeout: Duration, + surface: RpcSurface, ) -> io::Result<()> { stream.set_write_timeout(Some(idle_timeout))?; let mut reader = BufReader::new(RequestDeadlineStream::new(stream, idle_timeout)); @@ -327,7 +394,7 @@ fn serve_connection( Ok(Some(head)) => head, Ok(None) => return Ok(()), Err(RequestReadError::RestBodyTooLarge) => { - let response = if rest_enabled { + let response = if rest_enabled && surface != RpcSurface::Management { crate::rest::getutxos_body_too_large() } else { crate::rest::not_found() @@ -354,6 +421,18 @@ fn serve_connection( return Err(error); } }; + let route = classify(&head.method, &head.path); + let _request_timer = RequestTimer { + started: Instant::now(), + surface, + route: route.label(), + }; + // Reject a route outside this listener before authentication or body + // upload. Closing also prevents pipelined bytes crossing the boundary. + if !surface.allows(&route) { + write_not_found(reader.get_mut().socket_mut(), false)?; + return Ok(()); + } // Authentication runs from the head alone, before any body byte is // read: a 16 MiB declared body without credentials gets its 401 at // once instead of after its upload (Core `httprpc.cpp` checks inside @@ -366,7 +445,7 @@ fn serve_connection( let request = match read_request_body(&mut reader, REQUEST_BODY_TIMEOUT, head) { Ok(request) => request, Err(RequestReadError::RestBodyTooLarge) => { - let response = if rest_enabled { + let response = if rest_enabled && surface != RpcSurface::Management { crate::rest::getutxos_body_too_large() } else { crate::rest::not_found() @@ -1033,6 +1112,33 @@ enum HttpRoute<'a> { JsonRpc, } +impl HttpRoute<'_> { + const fn label(&self) -> &'static str { + match self { + Self::Rest { .. } | Self::RestPost { .. } => "rest", + Self::EsploraGet { + surface: HttpSurface::EsploraPublic, + .. + } + | Self::EsploraPost { + surface: HttpSurface::EsploraPublic, + .. + } + | Self::Preflight { .. } => "esplora_public", + Self::EsploraGet { + surface: HttpSurface::EsploraBackend, + .. + } + | Self::EsploraPost { + surface: HttpSurface::EsploraBackend, + .. + } => "esplora_backend", + Self::NotFound => "not_found", + Self::JsonRpc => "json_rpc", + } + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum HttpSurface { EsploraPublic, @@ -1209,6 +1315,8 @@ struct ResponseHead<'a> { #[cfg(test)] #[expect(clippy::expect_used)] mod tests { + mod listeners; + use super::*; use core::sync::atomic::{AtomicBool, Ordering}; @@ -1256,6 +1364,7 @@ mod tests { &Handler::new(Arc::new(Context::new())), false, Duration::from_secs(30), + RpcSurface::Shared, ) }); let started = Instant::now(); @@ -1504,6 +1613,7 @@ mod tests { &Handler::new(Arc::new(Context::new())), false, Duration::from_secs(30), + RpcSurface::Shared, ) }); let mut request = json_rpc_head(Some(GOOD_TOKEN), 16); @@ -1533,6 +1643,7 @@ mod tests { &handler, enabled, Duration::from_secs(2), + RpcSurface::Shared, ) }); client.write_all(request)?; @@ -1792,7 +1903,7 @@ mod tests { errors.push(io::Error::from_raw_os_error(10024)); // WSAEMFILE let mut errors = errors.into_iter(); let start = Instant::now(); - server.serve_accept_loop(&stop.into(), |listener| { + server.serve_accept_loop(&stop.into(), RpcSurface::Shared, |listener| { if let Some(error) = errors.next() { return Err(error); } @@ -1823,7 +1934,7 @@ mod tests { let stop = Arc::clone(&shutdown); let start = Instant::now(); let mut attempts = 0; - server.serve_accept_loop(&shutdown.into(), |_| { + server.serve_accept_loop(&shutdown.into(), RpcSurface::Shared, |_| { attempts += 1; if attempts == 3 { stop.store(true, Ordering::Release); @@ -1841,7 +1952,7 @@ mod tests { let server = accept_test_server()?; let shutdown = Arc::new(AtomicBool::new(false)); let mut attempts = 0; - let result = server.serve_accept_loop(&shutdown.into(), |_| { + let result = server.serve_accept_loop(&shutdown.into(), RpcSurface::Shared, |_| { attempts += 1; Err(kind.into()) }); @@ -2125,6 +2236,7 @@ mod tests { &handler, false, core::time::Duration::from_secs(5), + RpcSurface::Shared, ) }); @@ -2160,6 +2272,7 @@ mod tests { &handler, false, core::time::Duration::from_secs(5), + RpcSurface::Shared, ) }); @@ -2239,6 +2352,7 @@ mod tests { &handler, false, core::time::Duration::from_secs(1), + RpcSurface::Shared, ) }); diff --git a/crates/rpc/src/server/tests/listeners.rs b/crates/rpc/src/server/tests/listeners.rs new file mode 100644 index 000000000..b2d55522a --- /dev/null +++ b/crates/rpc/src/server/tests/listeners.rs @@ -0,0 +1,215 @@ +//! Independent admission and route boundaries for split HTTP listeners. + +use super::*; + +struct RunningServer { + address: SocketAddr, + shutdown: Arc, + worker: Option>>, + handler: Arc, +} + +impl RunningServer { + fn start(context: &Arc, surface: RpcSurface, limit: usize) -> io::Result { + let server = RpcServer::bind( + "127.0.0.1:0", + Arc::new(Auth::basic("alice", "secret")), + Arc::new(Handler::new(Arc::clone(context))), + limit, + Duration::from_secs(5), + true, + )?; + let address = server.local_addr()?; + let handler = Arc::clone(&server.handler); + let shutdown = Arc::new(AtomicBool::new(false)); + let stop = Arc::clone(&shutdown); + let worker = Some(thread::spawn(move || { + server.serve_surface_with_shutdown(stop, surface) + })); + Ok(Self { + address, + shutdown, + worker, + handler, + }) + } + + fn connect(&self) -> io::Result { + let stream = TcpStream::connect(self.address)?; + stream.set_read_timeout(Some(Duration::from_secs(3)))?; + stream.set_write_timeout(Some(Duration::from_secs(3)))?; + Ok(stream) + } + + fn request(&self, request: &[u8]) -> io::Result { + let mut stream = self.connect()?; + stream.write_all(request)?; + read_response(&mut stream) + } + + fn stop(&mut self) { + self.shutdown.store(true, Ordering::Release); + if let Some(worker) = self.worker.take() { + worker + .join() + .expect("listener thread") + .expect("listener exit"); + } + } +} + +impl Drop for RunningServer { + fn drop(&mut self) { + self.stop(); + } +} + +fn read_response(stream: &mut TcpStream) -> io::Result { + let mut reader = BufReader::new(stream); + let mut response = String::new(); + let mut length = 0; + loop { + let mut line = String::new(); + if reader.read_line(&mut line)? == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "missing HTTP head", + )); + } + if let Some(value) = line.strip_prefix("Content-Length: ") { + length = value.trim().parse().expect("response length"); + } + let end = line == "\r\n"; + response.push_str(&line); + if end { + break; + } + } + let mut body = vec![0; length]; + reader.read_exact(&mut body)?; + response.push_str(&String::from_utf8_lossy(&body)); + Ok(response) +} + +fn management_request() -> Vec { + let body = br#"{"jsonrpc":"2.0","id":1,"method":"getblockcount","params":[]}"#; + let mut request = json_rpc_head(Some(GOOD_TOKEN), body.len()); + request.extend_from_slice(body); + request +} + +#[test] +fn public_saturation_preserves_management_and_recovers() -> io::Result<()> { + let context = Arc::new(Context::new()); + let mut public = RunningServer::start(&context, RpcSurface::Public, 2)?; + let management = RunningServer::start(&context, RpcSurface::Management, 2)?; + assert!(Arc::ptr_eq( + public.handler.context(), + management.handler.context() + )); + assert!(!Arc::ptr_eq(&public.handler, &management.handler)); + + // An answered public keep-alive and an unclassified, incomplete head + // consume both public permits, even though neither uses JSON-RPC auth. + let mut idle = public.connect()?; + idle.write_all( + b"GET /api/blocks/tip/height HTTP/1.1\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n", + )?; + let response = read_response(&mut idle)?; + assert!( + response.contains("Access-Control-Allow-Origin: *"), + "{response}" + ); + let mut slow = public.connect()?; + slow.write_all(b"G")?; + let busy = read_response(&mut public.connect()?)?; + assert!( + busy.starts_with("HTTP/1.1 503 Service Unavailable"), + "{busy}" + ); + + let response = management.request(&management_request())?; + assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}"); + assert!(response.contains("\"result\":0"), "{response}"); + let unauthorized = management.request(&json_rpc_head(None, 0))?; + assert_eq!(unauthorized, CORE_UNAUTHORIZED); + + // EOF releases admission; retry is bounded, without assuming a worker + // scheduling delay. The incomplete head keeps the other public slot. + drop(idle); + let deadline = Instant::now() + Duration::from_secs(3); + loop { + let mut recovered = public.connect()?; + recovered.write_all( + b"GET /api/blocks/tip/height HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + )?; + let response = read_response(&mut recovered)?; + // Context::new has no applied tip, so the route itself can also + // return 503; CORS distinguishes a routed response from busy. + if response.contains("Access-Control-Allow-Origin: *") { + break; + } + assert!(Instant::now() < deadline, "public permit was not released"); + } + drop(slow); + public.stop(); + assert!(public.handler.is_stopping()); + assert!(!management.handler.is_stopping()); + assert!( + management + .request(&management_request())? + .contains("\"result\":0") + ); + // Management has its own finite pre-authentication bound too. + let _management_idle = management.connect()?; + let _management_slow = management.connect()?; + let busy = read_response(&mut management.connect()?)?; + assert!( + busy.starts_with("HTTP/1.1 503 Service Unavailable"), + "{busy}" + ); + Ok(()) +} + +#[test] +fn split_listener_routes_are_closed_before_authentication_and_body_upload() -> io::Result<()> { + let context = Arc::new(Context::new()); + let public = RunningServer::start(&context, RpcSurface::Public, 8)?; + let management = RunningServer::start(&context, RpcSurface::Management, 8)?; + for path in ["/", "/tx", "/unknown"] { + // Neither credentials nor an unuploaded body can reach JSON-RPC on + // the public listener. Legacy unknown POST fallback is closed too. + let request = format!( + "POST {path} HTTP/1.1\r\nAuthorization: Basic {GOOD_TOKEN}\r\nContent-Length: 4096\r\n\r\n" + ); + assert!( + public + .request(request.as_bytes())? + .starts_with("HTTP/1.1 404 Not Found") + ); + } + for path in ["/api/tx", "/esplora/internal/txs", "/rest/getutxos.bin"] { + let request = format!("POST {path} HTTP/1.1\r\nContent-Length: 4096\r\n\r\n"); + assert!( + management + .request(request.as_bytes())? + .starts_with("HTTP/1.1 404 Not Found") + ); + } + let preflight = public + .request(b"OPTIONS /api/tx HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")?; + assert!(preflight.starts_with("HTTP/1.1 204 No Content")); + assert!(preflight.contains("Access-Control-Allow-Origin: *")); + for path in [ + "/api/internal/txs", + "/esplora/unknown", + "/rest/unknown.json", + ] { + let response = public.request( + format!("GET {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .as_bytes(), + )?; + assert!(response.starts_with("HTTP/1.1 404 Not Found"), "{response}"); + } + Ok(()) +} diff --git a/docs/contracts/embedding.md b/docs/contracts/embedding.md index e6530104a..3ef84db33 100644 --- a/docs/contracts/embedding.md +++ b/docs/contracts/embedding.md @@ -12,7 +12,7 @@ the first embedder — there is one lifecycle implementation, not two. an RPC context for a caller to assemble. `StartupGuard::finish` transfers ownership only after every service has started. Both callers stop through `lifecycle.rs::NodeServices::teardown`: request shutdown and wake - the event loop; join the event loop and RPC listener; stop metrics; join + the event loop; join the event loop and all configured RPC/HTTP listeners; stop metrics; join P2P core, ingress, and relay workers; join bootstrap, checkpoint, and signal workers; then publish a clean checkpoint if eligible. On every stop path — including `StartupGuard` rollback — the diff --git a/docs/contracts/external-api.md b/docs/contracts/external-api.md index 23c9522b3..d7ab764eb 100644 --- a/docs/contracts/external-api.md +++ b/docs/contracts/external-api.md @@ -1046,6 +1046,24 @@ evidence does not establish mainnet memory or throughput qualifications. `busy`. A source is an address, so clients behind loopback, a Docker gateway or a reverse proxy share one. Pools below four workers keep no per-source share. + The node defaults to one shared listener with 128 connections. Opting into + `public_bind` moves the HTTP query/broadcast surfaces to an independently + bounded listener and leaves authenticated JSON-RPC on `rpc_bind` (WF-02). + `rpc_max_connections` and `public_max_connections` independently default to + 128 and must be positive. All admitted connections, including incomplete + heads, idle keep-alives and blocking requests, count against their binding. + Public occupancy cannot consume management permits; neither binding is + exempt from admission. Each accept loop can additionally hold one refused + socket while writing its 503, bounded by the configured idle timeout. + This isolates connection capacity, not shared CPU, process descriptors, + storage or lock contention. Each binding retains the + existing dispatcher-owned blocking quota and shutdown latch. Both listeners + bind before either starts; startup rollback and normal teardown own both. + Proof: `rpc::server::tests::listeners::public_saturation_preserves_management_and_recovers`, + `split_listener_routes_are_closed_before_authentication_and_body_upload`, + and node lifecycle tests + `split_rpc_listeners_start_and_join_through_node_services` and + `public_bind_failure_rolls_back_management_binding_and_node_state`. - Interrupted and aborted accepts retry immediately. Other accept errors, including descriptor exhaustion and Linux's pending-connection `EOPNOTSUPP` (`Unsupported`), warn and retry after 100 ms; shutdown is checked before diff --git a/docs/contracts/wallet-facing.md b/docs/contracts/wallet-facing.md index 8e8f3bdfc..9545735c3 100644 --- a/docs/contracts/wallet-facing.md +++ b/docs/contracts/wallet-facing.md @@ -29,8 +29,23 @@ stay in the external process. ### `WF-02`: Operations a wallet actually issues -Esplora lives at `/api` on the JSON-RPC listener. That directory is the -electrs/mempool.space base URL. Relative routes below are appended to it. +By default, Esplora lives at `/api` on the JSON-RPC listener. That directory +is the electrs/mempool.space base URL; relative routes below are appended to it. +With the opt-in +`public_bind`, REST (if enabled), `/api`, and `/esplora` move to that separate +public listener, and `rpc_bind` serves authenticated JSON-RPC only. The public +listener refuses JSON-RPC, including the legacy unknown-POST fallback, with +404 and closes before authentication or body upload. Both bindings share the +same Context and authoritative services; routing, capability gates, broadcast +origins and CORS inside the public namespaces are unchanged. Each binding +owns its connection budget, dispatcher cancellation latch and blocking quota +(API-34). Proof: +`rpc::server::tests::listeners::split_listener_routes_are_closed_before_authentication_and_body_upload`, +`rpc::server::tests::listeners::public_saturation_preserves_management_and_recovers`, +and node lifecycle tests +`split_rpc_listeners_start_and_join_through_node_services` and +`public_bind_failure_rolls_back_management_binding_and_node_state`. + Every read captures the applied-tip publication and answers from one coherent view. A moved chain generation returns the declared unavailable response, never a mixed-tip page. diff --git a/docs/getting-started.md b/docs/getting-started.md index 44ce9e921..975dd63d4 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -180,11 +180,36 @@ Example regtest startup: The wallet-facing contract is [contracts/wallet-facing.md](contracts/wallet-facing.md). +To keep public connections from using management RPC capacity, add +`--public-bind 0.0.0.0:8080`. In this opt-in mode `/api`, `/esplora`, and +REST (when enabled) move to the public port; `--rpc-bind` serves authenticated +JSON-RPC only. Keep the management port on loopback or a trusted network. +The public port preserves both existing Esplora namespaces, including backend +helpers and transaction broadcast; choose exposure accordingly. A public port +does not accept JSON-RPC, even with valid credentials. + +`--rpc-max-connections` and `--public-max-connections` each default to 128 +and must be positive. The bounds include slow/incomplete requests and idle +keep-alives, with the existing per-source reservation and head/body/idle +timeouts. Separation protects connection capacity; CPU, process descriptor +limits, storage and locks remain shared. Without `--public-bind`, the existing +combined listener and routes remain the default. + +These options also resolve through TOML (`public_bind`, `rpc_max_connections`, +`public_max_connections`) and environment variables (`BITCOIN_RS_PUBLIC_BIND`, +`BITCOIN_RS_RPC_MAX_CONNECTIONS`, `BITCOIN_RS_PUBLIC_MAX_CONNECTIONS`), using +the normal file < environment < CLI precedence. Both listeners use the same +chain, mempool, indexes and broadcast gateway. + A Compose example runs the upstream mempool explorer against these endpoints: [btc-mempool-explorer](../tools/btc-mempool-explorer/README.md). ### REST -`--rest=true` (or `rest=1` in `bitcoin.conf`) serves Core-compatible REST on the RPC port without authentication. The CLI option takes an explicit boolean; bare `--rest` is not the enablement syntax. See [rest-interface.md](rest-interface.md). +`--rest=true` (or `rest=1` in `bitcoin.conf`) serves Core-compatible REST without +authentication on the combined `--rpc-bind` listener by default. With +`--public-bind`, REST moves to that public listener and the RPC port serves +management JSON-RPC only. The CLI option takes an explicit boolean; bare +`--rest` is not the enablement syntax. See [rest-interface.md](rest-interface.md). ### ZMQ diff --git a/docs/observability.md b/docs/observability.md index ee955e6ca..b3f4299fc 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -37,6 +37,22 @@ Benchmark evidence identity (binary/configuration/corpus digests and hardware) is recorded with benchmark samples under HPA-12; it is not attached to every Prometheus series as global labels. +RPC connection isolation (API-34/WF-02) exposes `rpc.connections.active` +(gauge, admitted live workers), `rpc.connections.rejected_total` (counter, +connection-budget refusals), and `rpc.http.request_seconds` (histogram, +complete request head through body read, authentication and response write, +including subsequent failures; excludes idle/head acquisition). The closed `surface` +label is `shared`, `management`, or `public`; the histogram's closed `route` +label is `json_rpc`, `rest`, `esplora_public`, `esplora_backend`, or `not_found`. +`route` records classification before the listener's route policy is checked; +it includes refused routes. For example, `surface=management, route=rest` +records a REST request rejected by the management listener with 404. +Values aggregate bindings of the same surface. Operators can distinguish +public saturation from management pressure and compare classified request +latency. Occupancy updates follow the existing connection permit lifetime, +including spawn failure and shutdown; no second admission counter is added. +No paths, method names, credentials, addresses, hashes or txids are labels. + ### `OBS-02`: `tracing::` is diagnostics, not an API Free-form development and debugging detail belongs in `tracing::` events and diff --git a/docs/rest-interface.md b/docs/rest-interface.md index c8c376e80..cafe81149 100644 --- a/docs/rest-interface.md +++ b/docs/rest-interface.md @@ -1,8 +1,9 @@ # REST interface bitcoin-rs can expose the small Bitcoin Core-compatible REST surface needed by -remote chain validators. REST uses the existing JSON-RPC listener and port; it -does not create a second listener. Enable it on the CLI (`--rest=true`), via +remote chain validators. REST uses the existing JSON-RPC listener and port by +default. With `--public-bind`, REST moves to that separate public listener; +see [public consumers](getting-started.md#public-consumers). Enable it on the CLI (`--rest=true`), via `BITCOIN_RS_REST=true`, in TOML (`rest = true`), or with Core-style `bitcoin.conf` configuration: @@ -10,10 +11,11 @@ does not create a second listener. Enable it on the CLI (`--rest=true`), via rest=1 ``` -The REST requests are unauthenticated, as in Bitcoin Core. JSON-RPC requests on -the same listener continue to require their configured authentication. Select -the listener with the existing `--rpc-bind` option (or its layered config -equivalent; see [getting-started.md](getting-started.md#rest)). +The REST requests are unauthenticated, as in Bitcoin Core. JSON-RPC requests +continue to require their configured authentication. By default, `--rpc-bind` +selects the combined listener. In split mode, `--public-bind` selects the REST +listener and `--rpc-bind` serves management JSON-RPC only (or use the layered +config equivalents; see [getting-started.md](getting-started.md#rest)). The gateway registers these Core REST prefixes: diff --git a/tools/stratum-sv2/template-provider/Cargo.lock b/tools/stratum-sv2/template-provider/Cargo.lock index 063d3fcc5..1a960f55f 100644 --- a/tools/stratum-sv2/template-provider/Cargo.lock +++ b/tools/stratum-sv2/template-provider/Cargo.lock @@ -426,6 +426,7 @@ dependencies = [ "corepc-types", "crossbeam-channel", "hashbrown 0.17.1", + "metrics", "miniscript", "parking_lot", "serde",