SAFE21.io/bal-server
1
0
Fork 0
forked from bitcoinafterlife/bal-server
Code Pull Requests Activity

fix: per-endpoint rate limits, request timeout and input hardening

Browse Source 
bal-server:
- Wire independent actix-governor buckets per endpoint class (pushtxs,
  searchtx, info/stats, default) so polling a cheap endpoint can no longer
  exhaust the push-txs budget; previously every limit was dead config
- Enforce BAL_SERVER_ACTIX_TIMEOUT_SECS via a request-timeout middleware that
  returns 408 Request Timeout, covering slow request bodies
- De-duplicate txids within a single /pushtxs request body
- Replace the byte-slice xpub detection (panic on non-ASCII) with a safe
  prefix check

bal-pusher:
- Handle malformed/empty ZMQ frames without indexing panics or UTF-8 expect
- Return an error from sign_message and skip the stats report instead of
  panicking when the private key cannot be parsed

validation / db:
- Reject IPv4-mapped IPv6 targets (::ffff:a.b.c.d) in is_valid_welist_url
  to close an SSRF bypass, with unit tests
- Use Path::starts_with for the SQLite forbidden-directory check so paths
  like /etcfoo are not misclassified

misc:
- Add explicit secret patterns to .gitignore and sync the docs
This commit is contained in:
svatantrya
2026-10-14 18:30:23 -04:00
parent 457c143954
commit b8d87ecabd
8 changed files with 224 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/bin/bal-pusher.rs
View File

@@ -388,7 +388,13 @@ async fn send_stats_report(
cfg.url, chain, bcinfo.blocks, bcinfo.median_time, bcinfo.best_block_hash
);
trace!("message to be sent: {}", message);
let sign = sign_message(cfg.ssl_key_path.as_str(), message.as_str());
let sign = match sign_message(cfg.ssl_key_path.as_str(), message.as_str()) {
Ok(signature) => signature,
Err(e) => {
error!("Cannot sign stats report, skipping: {}", e);
return Ok(());
}
};
let response = client
.post(url)
.header("User-Agent", format!("bal-pusher/{}", VERSION))
@@ -414,12 +420,12 @@ async fn send_stats_report(
}
Ok(())
}
fn sign_message(private_key_path: &str, message: &str) -> String {
let signing_key =
SigningKey::read_pkcs8_pem_file(private_key_path).expect("failed to parse private key PEM");
fn sign_message(private_key_path: &str, message: &str) -> Result<String, Box<dyn StdError>> {
let signing_key = SigningKey::read_pkcs8_pem_file(private_key_path)
.map_err(|e| format!("failed to parse private key PEM at {private_key_path}: {e}"))?;
let signature = signing_key.sign(message.as_bytes());
general_purpose::STANDARD.encode(signature.to_bytes())
Ok(general_purpose::STANDARD.encode(signature.to_bytes()))
}
fn parse_env(cfg: &mut MyConfig) {
@@ -561,12 +567,12 @@ async fn main() -> std::io::Result<()> {
);
}
consecutive_timeouts = 0;
let topic = message[0].clone();
let body = message[1].clone();
debug!(
"ZMQ:GET TOPIC: {}",
String::from_utf8(topic.clone()).expect("invalid topic")
);
let Some(topic) = message.first().cloned() else {
warn!("ZMQ: received empty multipart message, skipping");
continue;
};
let body = message.get(1).cloned().unwrap_or_default();
debug!("ZMQ:GET TOPIC: {}", String::from_utf8_lossy(&topic));
trace!("ZMQ:GET BODY: {}", hex::encode(&body));
if topic == b"hashblock" {
info!("NEW BLOCK: {}", hex::encode(&body));

132
src/bin/bal-server.rs
View File

@@ -1,8 +1,11 @@
use actix_governor::{Governor, GovernorConfigBuilder, KeyExtractor, SimpleKeyExtractionError};
use actix_web::dev::ServiceRequest;
use actix_web::middleware;
use actix_governor::{
Governor, GovernorConfig, GovernorConfigBuilder, KeyExtractor, SimpleKeyExtractionError,
};
use actix_web::body::MessageBody;
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::middleware::{self, Next};
use actix_web::web::Bytes;
use actix_web::{App, HttpResponse, HttpServer, Responder, web};
use actix_web::{App, Error, HttpResponse, HttpServer, Responder, web};
use bitcoin::{Network, Transaction, consensus};
use chrono::Utc;
use hex_conservative::FromHex;
@@ -13,6 +16,7 @@ use std::env;
use std::fs;
use std::net::IpAddr;
use std::str::FromStr;
use std::time::Duration;
use bal_server::db::{
DatabasePool, InsertInpData, InsertOutData, InsertTxData, check_duplicate_txids,
@@ -123,7 +127,6 @@ pub struct StatsResponse {
}
#[derive(Debug, Clone)]
#[expect(dead_code)]
struct ActixConfig {
max_body_size: usize,
timeout_secs: u64,
@@ -287,6 +290,44 @@ impl KeyExtractor for RealIpKeyExtractor {
}
}
/// Concrete governor configuration type used by all rate-limited routes.
///
/// `NoOpMiddleware` is the default `actix-governor` middleware; it is only used
/// here as a type parameter so a single helper can build every route's config.
type GovernorConfigType =
GovernorConfig<RealIpKeyExtractor, actix_governor::governor::middleware::NoOpMiddleware>;
/// Build a rate-limiter configuration keyed by the real client IP.
///
/// A period of zero is rejected by `actix-governor`, so the values are clamped
/// to a minimum of one to avoid a `None` result.
fn build_governor_config(per_sec: u64, burst: u32) -> Result<GovernorConfigType, String> {
GovernorConfigBuilder::default()
.seconds_per_request(per_sec.max(1))
.burst_size(burst.max(1))
.key_extractor(RealIpKeyExtractor)
.finish()
.ok_or_else(|| "invalid rate limiter configuration".to_string())
}
/// Middleware that aborts a request with `408 Request Timeout` if the inner
/// handler (including request body reading) exceeds the configured duration.
///
/// The duration is read from the `web::Data<u64>` app data registered in `main`.
async fn request_timeout<B>(
timeout_secs: web::Data<u64>,
req: ServiceRequest,
next: Next<B>,
) -> Result<ServiceResponse<B>, Error>
where
B: MessageBody,
{
match tokio::time::timeout(Duration::from_secs(**timeout_secs), next.call(req)).await {
Ok(result) => result,
Err(_) => Err(actix_web::error::ErrorRequestTimeout("request timeout")),
}
}
async fn echo_info(
path: web::Path<String>,
data: web::Data<AppState>,
@@ -464,6 +505,7 @@ fn parse_request_transactions(
known_addresses: &HashSet<String>,
) -> Vec<(ParsedTx, String, u64)> {
let mut result: Vec<(ParsedTx, String, u64)> = Vec::new();
let mut seen_txids: HashSet<String> = HashSet::new();
for line in strbody.split('\n') {
if line.is_empty() {
@@ -489,6 +531,10 @@ fn parse_request_transactions(
};
let txid = tx.compute_txid().to_string();
if !seen_txids.insert(txid.clone()) {
trace!("duplicate txid {} in request body, skipping", txid);
continue;
}
let ntxid = tx.compute_ntxid();
let wtxid = tx.compute_wtxid();
let locktime = tx.lock_time.to_string();
@@ -750,7 +796,10 @@ fn parse_env_netconfig(cfg: &mut MyConfig, chain: &str) {
if let Ok(value) = env::var(format!("BAL_SERVER_{}_ADDRESS", chain.to_uppercase())) {
debug!("BAL_SERVER_{}_ADDRESS: {}", chain.to_uppercase(), value);
c.address = value;
if c.address.len() > 5 && &c.address[1..4] == "pub" {
let is_xpub = ["xpub", "ypub", "zpub", "tpub", "vpub", "upub"]
.iter()
.any(|prefix| c.address.starts_with(prefix));
if is_xpub {
c.xpub = true;
trace!("is_xpub");
}
@@ -813,12 +862,26 @@ async fn main() -> std::io::Result<()> {
let bind_address = data.cfg.bind_address.clone();
let bind_port = data.cfg.bind_port;
let governor_conf = GovernorConfigBuilder::default()
.seconds_per_request(actix_cfg.rate_limit_pushtxs.0)
.burst_size(actix_cfg.rate_limit_pushtxs.1)
.key_extractor(RealIpKeyExtractor)
.finish()
.unwrap();
// One shared limiter per endpoint class: all workers clone the same
// underlying state, so the rate limit is global per real client IP.
let governor_pushtxs = build_governor_config(
actix_cfg.rate_limit_pushtxs.0,
actix_cfg.rate_limit_pushtxs.1,
)
.map_err(std::io::Error::other)?;
let governor_searchtx = build_governor_config(
actix_cfg.rate_limit_searchtx.0,
actix_cfg.rate_limit_searchtx.1,
)
.map_err(std::io::Error::other)?;
let governor_info =
build_governor_config(actix_cfg.rate_limit_info.0, actix_cfg.rate_limit_info.1)
.map_err(std::io::Error::other)?;
let governor_default = build_governor_config(
actix_cfg.rate_limit_default.0,
actix_cfg.rate_limit_default.1,
)
.map_err(std::io::Error::other)?;
println!("Starting server on http://{}:{}", bind_address, bind_port);
@@ -827,16 +890,45 @@ async fn main() -> std::io::Result<()> {
.app_data(web::PayloadConfig::default().limit(actix_cfg.max_body_size))
.app_data(data.clone())
.app_data(web::Data::new(actix_cfg.trusted_proxy))
.app_data(web::Data::new(actix_cfg.timeout_secs))
.wrap(middleware::Logger::default())
.wrap(middleware::Compress::default())
.wrap(Governor::new(&governor_conf))
.service(web::resource("/").route(web::get().to(echo_home)))
.service(web::resource("/.pub_key.pem").route(web::get().to(echo_pub_key)))
.service(web::resource("/version").route(web::get().to(echo_version)))
.service(web::resource("/{network}/info").route(web::get().to(echo_info)))
.service(web::resource("/{network}/stats").route(web::get().to(echo_stats)))
.service(web::resource("/{network}/pushtxs").route(web::post().to(echo_push)))
.service(web::resource("/searchtx").route(web::post().to(echo_search)))
.wrap(middleware::from_fn(request_timeout))
.service(
web::resource("/")
.wrap(Governor::new(&governor_default))
.route(web::get().to(echo_home)),
)
.service(
web::resource("/.pub_key.pem")
.wrap(Governor::new(&governor_default))
.route(web::get().to(echo_pub_key)),
)
.service(
web::resource("/version")
.wrap(Governor::new(&governor_default))
.route(web::get().to(echo_version)),
)
.service(
web::resource("/{network}/info")
.wrap(Governor::new(&governor_info))
.route(web::get().to(echo_info)),
)
.service(
web::resource("/{network}/stats")
.wrap(Governor::new(&governor_info))
.route(web::get().to(echo_stats)),
)
.service(
web::resource("/{network}/pushtxs")
.wrap(Governor::new(&governor_pushtxs))
.route(web::post().to(echo_push)),
)
.service(
web::resource("/searchtx")
.wrap(Governor::new(&governor_searchtx))
.route(web::post().to(echo_search)),
)
})
.workers(actix_cfg.workers)
.max_connections(actix_cfg.max_connections)

16
src/db/mod.rs
View File

@@ -38,13 +38,23 @@ fn validate_sqlite_path(dsn: &str) -> Result<(), String> {
// If absolute, block known sensitive system directories
if p.is_absolute() {
let forbidden = [
"/etc", "/proc", "/sys", "/dev", "/usr", "/bin", "/sbin", "/lib", "/opt",
Path::new("/etc"),
Path::new("/proc"),
Path::new("/sys"),
Path::new("/dev"),
Path::new("/usr"),
Path::new("/bin"),
Path::new("/sbin"),
Path::new("/lib"),
Path::new("/opt"),
];
for prefix in &forbidden {
if path_str.starts_with(prefix) {
// `Path::starts_with` matches whole path components, so "/etcfoo"
// is not treated as being under "/etc".
if p.starts_with(prefix) {
return Err(format!(
"Absolute database path under {} is forbidden",
prefix
prefix.display()
));
}
}

45
src/validation.rs
View File

@@ -65,6 +65,19 @@ pub fn is_valid_welist_url(url_str: &str) -> bool {
{
return false;
}
// IPv4-mapped IPv6 addresses (e.g. `::ffff:127.0.0.1`) are
// routed to the embedded IPv4 address on dual-stack hosts, so
// they must be validated against the IPv4 rules as well.
if let Some(v4) = v6.to_ipv4_mapped()
&& (v4.is_loopback()
|| v4.is_private()
|| v4.is_link_local()
|| v4.is_unspecified()
|| v4.is_multicast())
{
return false;
}
}
}
}
@@ -154,6 +167,38 @@ mod tests {
);
}
#[test]
fn test_ipv4_mapped_ipv6() {
assert!(
!is_valid_welist_url("https://[::ffff:127.0.0.1]"),
"IPv4-mapped loopback should be blocked"
);
assert!(
!is_valid_welist_url("https://[::ffff:10.0.0.1]"),
"IPv4-mapped private address should be blocked"
);
assert!(
!is_valid_welist_url("https://[::ffff:172.16.0.1]"),
"IPv4-mapped private address should be blocked"
);
assert!(
!is_valid_welist_url("https://[::ffff:192.168.1.1]"),
"IPv4-mapped private address should be blocked"
);
assert!(
!is_valid_welist_url("https://[::ffff:169.254.169.254]"),
"IPv4-mapped link-local address should be blocked"
);
assert!(
!is_valid_welist_url("https://[::ffff:0.0.0.0]"),
"IPv4-mapped unspecified address should be blocked"
);
assert!(
is_valid_welist_url("https://[::ffff:8.8.8.8]"),
"IPv4-mapped public address should be allowed"
);
}
#[test]
fn test_malformed_urls() {
assert!(!is_valid_welist_url("not a url"));