Compare commits
5 Commits
22b60e55c7
...
0f0f0a08c3
| Author | SHA1 | Date | |
|---|---|---|---|
|
0f0f0a08c3
|
|||
|
6e6c634e98
|
|||
|
efcd91e6b4
|
|||
|
cd24eda111
|
|||
|
8ce3f6a445
|
3
.gitignore
vendored
3
.gitignore
vendored
@@ -8,9 +8,6 @@
|
|||||||
.env.production
|
.env.production
|
||||||
.env.secret
|
.env.secret
|
||||||
|
|
||||||
# Shell scripts that load env vars (contain secrets, local only)
|
|
||||||
bal-pusher.sh
|
|
||||||
bal-server.sh
|
|
||||||
|
|
||||||
# Private keys - NEVER commit to git
|
# Private keys - NEVER commit to git
|
||||||
# Only public_key.pem should be tracked (if needed)
|
# Only public_key.pem should be tracked (if needed)
|
||||||
|
|||||||
2
Cargo.lock
generated
2
Cargo.lock
generated
@@ -312,7 +312,7 @@ checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "bal_server"
|
name = "bal_server"
|
||||||
version = "0.3.0"
|
version = "0.3.1"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"actix-governor",
|
"actix-governor",
|
||||||
"actix-rt",
|
"actix-rt",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "bal_server"
|
name = "bal_server"
|
||||||
version = "0.3.0"
|
version = "0.3.1"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
|
|
||||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||||
|
|||||||
15
bal-pusher.sh
Normal file
15
bal-pusher.sh
Normal file
@@ -0,0 +1,15 @@
|
|||||||
|
export RUST_LOG=trace
|
||||||
|
|
||||||
|
export BAL_PUSHER_DB_FILE="$(pwd)/bal.db"
|
||||||
|
#export BAL_PUSHER_BITCOIN_COOKIE_FILE=/~/.bitcoin/.cookie
|
||||||
|
#export BAL_PUSHER_REGTEST_COOKIE_FILE=/~/.bitcoin/regtest/.cookie
|
||||||
|
#export BAL_PUSHER_TESTNET_COOKIE_FILE=/~/.bitcoin/testnet3/.cookie
|
||||||
|
#export BAL_PUSHER_SIGNET_COOKIE_FILE=/~/.bitcoin/signet/.cookie
|
||||||
|
|
||||||
|
export BAL_PUSHER_REGTEST_ZMQ_HASHBLOCK=tcp://127.0.0.1:21332
|
||||||
|
export BAL_PUSHER_SEND_STATS=true
|
||||||
|
export WELIST_SERVER_URL=http://localhost:8086
|
||||||
|
export WELIST_SKIP_URL_VALIDATION=true
|
||||||
|
export BAL_SERVER_URL="http://127.0.0.1:9133"
|
||||||
|
export SSL_KEY_PATH="$(pwd)/private_key.pem"
|
||||||
|
cargo run --bin=bal-pusher regtest
|
||||||
@@ -6,7 +6,6 @@ use bitcoincore_rpc::{Auth, Client, Error, RpcApi, bitcoin};
|
|||||||
use bitcoincore_rpc_json::GetBlockchainInfoResult;
|
use bitcoincore_rpc_json::GetBlockchainInfoResult;
|
||||||
|
|
||||||
use byteorder::{LittleEndian, ReadBytesExt};
|
use byteorder::{LittleEndian, ReadBytesExt};
|
||||||
use hex;
|
|
||||||
use log::{debug, error, info, trace, warn};
|
use log::{debug, error, info, trace, warn};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
@@ -23,10 +22,8 @@ use zmq::{Context, DEALER, DONTWAIT, Socket};
|
|||||||
use bal_server::db::open_db;
|
use bal_server::db::open_db;
|
||||||
use bal_server::validation::is_valid_welist_url;
|
use bal_server::validation::is_valid_welist_url;
|
||||||
use base64::{Engine as _, engine::general_purpose};
|
use base64::{Engine as _, engine::general_purpose};
|
||||||
use openssl::hash::MessageDigest;
|
|
||||||
use openssl::pkey::PKey;
|
use openssl::pkey::PKey;
|
||||||
use openssl::sign::Signer;
|
use openssl::sign::Signer;
|
||||||
use openssl::sign::Verifier;
|
|
||||||
use reqwest::Client as rClient;
|
use reqwest::Client as rClient;
|
||||||
use std::fs;
|
use std::fs;
|
||||||
use std::net::SocketAddr;
|
use std::net::SocketAddr;
|
||||||
@@ -145,7 +142,7 @@ fn get_network_params_default(network: Network) -> NetworkParams {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn get_cookie_filename(network: &NetworkParams) -> Result<String, Box<dyn StdError>> {
|
fn get_cookie_filename(network: &NetworkParams) -> Result<String, Box<dyn StdError>> {
|
||||||
if network.cookie_file != "" {
|
if !network.cookie_file.is_empty() {
|
||||||
Ok(network.cookie_file.clone())
|
Ok(network.cookie_file.clone())
|
||||||
} else {
|
} else {
|
||||||
match env::var_os("HOME") {
|
match env::var_os("HOME") {
|
||||||
@@ -163,12 +160,12 @@ fn get_cookie_filename(network: &NetworkParams) -> Result<String, Box<dyn StdErr
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
fn get_client_from_username(
|
fn get_client_from_username(
|
||||||
url: &String,
|
url: &str,
|
||||||
network: &NetworkParams,
|
network: &NetworkParams,
|
||||||
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
|
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
|
||||||
if network.rpc_user != "" {
|
if !network.rpc_user.is_empty() {
|
||||||
match Client::new(
|
match Client::new(
|
||||||
&url[..],
|
url,
|
||||||
Auth::UserPass(network.rpc_user.to_string(), network.rpc_pass.to_string()),
|
Auth::UserPass(network.rpc_user.to_string(), network.rpc_pass.to_string()),
|
||||||
) {
|
) {
|
||||||
Ok(client) => match client.get_blockchain_info() {
|
Ok(client) => match client.get_blockchain_info() {
|
||||||
@@ -182,30 +179,30 @@ fn get_client_from_username(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
fn get_client_from_cookie(
|
fn get_client_from_cookie(
|
||||||
url: &String,
|
url: &str,
|
||||||
network: &NetworkParams,
|
network: &NetworkParams,
|
||||||
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
|
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
|
||||||
match get_cookie_filename(network) {
|
match get_cookie_filename(network) {
|
||||||
Ok(cookie) => match Client::new(&url[..], Auth::CookieFile(cookie.into())) {
|
Ok(cookie) => match Client::new(url, Auth::CookieFile(cookie.into())) {
|
||||||
Ok(client) => match client.get_blockchain_info() {
|
Ok(client) => match client.get_blockchain_info() {
|
||||||
Ok(bcinfo) => Ok((client, bcinfo)),
|
Ok(bcinfo) => Ok((client, bcinfo)),
|
||||||
Err(err) => Err(err.into()),
|
Err(err) => Err(err.into()),
|
||||||
},
|
},
|
||||||
Err(err) => Err(err.into()),
|
Err(err) => Err(err.into()),
|
||||||
},
|
},
|
||||||
Err(err) => Err(err.into()),
|
Err(err) => Err(err),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
fn get_client(
|
fn get_client(
|
||||||
network: &NetworkParams,
|
network: &NetworkParams,
|
||||||
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
|
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
|
||||||
let url = format!("{}:{}/", network.host, &network.port);
|
let url = format!("{}:{}/", network.host, network.port);
|
||||||
debug!("trying to connect to bitcoin daemon:{url}");
|
debug!("trying to connect to bitcoin daemon:{url}");
|
||||||
match get_client_from_username(&url, network) {
|
match get_client_from_username(&url, network) {
|
||||||
Ok(client) => Ok(client),
|
Ok(client) => Ok(client),
|
||||||
Err(_) => match get_client_from_cookie(&url, &network) {
|
Err(_) => match get_client_from_cookie(&url, network) {
|
||||||
Ok(client) => Ok(client),
|
Ok(client) => Ok(client),
|
||||||
Err(err) => Err(err.into()),
|
Err(err) => Err(err),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -267,7 +264,7 @@ async fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(
|
|||||||
let mut invalid_txs: std::collections::HashMap<String, String> = HashMap::new();
|
let mut invalid_txs: std::collections::HashMap<String, String> = HashMap::new();
|
||||||
for row_result in match query_tx.bind::<&[(_, Value)]>(
|
for row_result in match query_tx.bind::<&[(_, Value)]>(
|
||||||
&[
|
&[
|
||||||
(":locktime_threshold", (LOCKTIME_THRESHOLD as i64).into()),
|
(":locktime_threshold", LOCKTIME_THRESHOLD.into()),
|
||||||
(":bestblock_time", (average_time as i64).into()),
|
(":bestblock_time", (average_time as i64).into()),
|
||||||
(":bestblock_height", (bcinfo.blocks as i64).into()),
|
(":bestblock_height", (bcinfo.blocks as i64).into()),
|
||||||
(":network", network_params.db_field.clone().into()),
|
(":network", network_params.db_field.clone().into()),
|
||||||
@@ -334,9 +331,7 @@ async fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(
|
|||||||
let _ = stmt.next();
|
let _ = stmt.next();
|
||||||
}
|
}
|
||||||
if let Err(e) = send_stats_report(cfg, bcinfo).await {
|
if let Err(e) = send_stats_report(cfg, bcinfo).await {
|
||||||
// Never discard silently: a failing report is otherwise
|
error!("send_stats_report failed: {}", e);
|
||||||
// invisible in the logs and can go unnoticed for a long time.
|
|
||||||
warn!("send_stats_report failed: {e}");
|
|
||||||
}
|
}
|
||||||
if let Err(e) = calculate_stats(&db, network_params.db_field.clone()).await {
|
if let Err(e) = calculate_stats(&db, network_params.db_field.clone()).await {
|
||||||
warn!("calculate_stats failed: {e}");
|
warn!("calculate_stats failed: {e}");
|
||||||
@@ -476,30 +471,38 @@ async fn welist_http_client(welist_url: &str) -> rClient {
|
|||||||
.parse::<bool>()
|
.parse::<bool>()
|
||||||
.unwrap_or(false);
|
.unwrap_or(false);
|
||||||
if !prefer_ipv6 {
|
if !prefer_ipv6 {
|
||||||
return rClient::new();
|
return new_welist_client();
|
||||||
}
|
}
|
||||||
let (host, port) = match parse_host_port(welist_url) {
|
let (host, port) = match parse_host_port(welist_url) {
|
||||||
Some(hp) => hp,
|
Some(hp) => hp,
|
||||||
None => {
|
None => {
|
||||||
warn!("BAL_PUSHER_PREFER_IPV6: cannot parse '{welist_url}', using default resolver");
|
warn!("BAL_PUSHER_PREFER_IPV6: cannot parse '{welist_url}', using default resolver");
|
||||||
return rClient::new();
|
return new_welist_client();
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
match resolve_first_ipv6(&host, port).await {
|
match resolve_first_ipv6(&host, port).await {
|
||||||
Some(addr) => {
|
Some(addr) => {
|
||||||
debug!("BAL_PUSHER_PREFER_IPV6: pinning {host} to {addr}");
|
debug!("BAL_PUSHER_PREFER_IPV6: pinning {host} to {addr}");
|
||||||
rClient::builder()
|
rClient::builder()
|
||||||
|
.timeout(Duration::from_secs(10))
|
||||||
.resolve(&host, addr)
|
.resolve(&host, addr)
|
||||||
.build()
|
.build()
|
||||||
.unwrap_or_else(|_| rClient::new())
|
.unwrap_or_else(|_| new_welist_client())
|
||||||
}
|
}
|
||||||
None => {
|
None => {
|
||||||
debug!("BAL_PUSHER_PREFER_IPV6: no IPv6 address for {host}, using default resolver");
|
debug!("BAL_PUSHER_PREFER_IPV6: no IPv6 address for {host}, using default resolver");
|
||||||
rClient::new()
|
new_welist_client()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn new_welist_client() -> rClient {
|
||||||
|
rClient::builder()
|
||||||
|
.timeout(Duration::from_secs(10))
|
||||||
|
.build()
|
||||||
|
.unwrap_or_else(|_| rClient::new())
|
||||||
|
}
|
||||||
|
|
||||||
async fn send_stats_report(
|
async fn send_stats_report(
|
||||||
cfg: &MyConfig,
|
cfg: &MyConfig,
|
||||||
bcinfo: GetBlockchainInfoResult,
|
bcinfo: GetBlockchainInfoResult,
|
||||||
@@ -508,7 +511,11 @@ async fn send_stats_report(
|
|||||||
debug!("sending report to welist");
|
debug!("sending report to welist");
|
||||||
let welist_url = env::var("WELIST_SERVER_URL")
|
let welist_url = env::var("WELIST_SERVER_URL")
|
||||||
.unwrap_or("https://welist.bitcoin-after.life".to_string());
|
.unwrap_or("https://welist.bitcoin-after.life".to_string());
|
||||||
if !is_valid_welist_url(&welist_url) {
|
let skip_validation = env::var("WELIST_SKIP_URL_VALIDATION")
|
||||||
|
.unwrap_or("false".to_string())
|
||||||
|
.parse::<bool>()
|
||||||
|
.unwrap_or(false);
|
||||||
|
if !skip_validation && !is_valid_welist_url(&welist_url) {
|
||||||
warn!(
|
warn!(
|
||||||
"Invalid or unsafe WELIST_SERVER_URL: {}. Skipping stats report.",
|
"Invalid or unsafe WELIST_SERVER_URL: {}. Skipping stats report.",
|
||||||
welist_url
|
welist_url
|
||||||
@@ -524,7 +531,7 @@ async fn send_stats_report(
|
|||||||
cfg.url, chain, bcinfo.blocks, bcinfo.median_time, bcinfo.best_block_hash
|
cfg.url, chain, bcinfo.blocks, bcinfo.median_time, bcinfo.best_block_hash
|
||||||
);
|
);
|
||||||
trace!("message to be sent: {}", message);
|
trace!("message to be sent: {}", message);
|
||||||
let sign = sign_message(cfg.ssl_key_path.as_str(), &message.as_str());
|
let sign = sign_message(cfg.ssl_key_path.as_str(), message.as_str());
|
||||||
let response = client
|
let response = client
|
||||||
.post(url)
|
.post(url)
|
||||||
.header("User-Agent", format!("bal-pusher/{}", VERSION))
|
.header("User-Agent", format!("bal-pusher/{}", VERSION))
|
||||||
@@ -539,16 +546,12 @@ async fn send_stats_report(
|
|||||||
}))
|
}))
|
||||||
.send()
|
.send()
|
||||||
.await?;
|
.await?;
|
||||||
if !response.status().is_success() {
|
let status = response.status();
|
||||||
warn!(
|
let body = response.text().await?;
|
||||||
"Non-success response: {} {}",
|
info!(
|
||||||
response.status(),
|
"Report to welist({}) status={} body={}",
|
||||||
response.status().canonical_reason().unwrap_or("")
|
welist_url, status, body
|
||||||
);
|
);
|
||||||
}
|
|
||||||
|
|
||||||
let body = &(response.text().await?);
|
|
||||||
info!("Report to welist({})\tSent: {}", welist_url, body);
|
|
||||||
} else {
|
} else {
|
||||||
debug!("Not sending stats");
|
debug!("Not sending stats");
|
||||||
}
|
}
|
||||||
@@ -562,9 +565,7 @@ fn sign_message(private_key_path: &str, message: &str) -> String {
|
|||||||
|
|
||||||
let signature = signer.sign_oneshot_to_vec(message.as_bytes()).unwrap();
|
let signature = signer.sign_oneshot_to_vec(message.as_bytes()).unwrap();
|
||||||
|
|
||||||
let signature_b64 = general_purpose::STANDARD.encode(&signature);
|
general_purpose::STANDARD.encode(&signature)
|
||||||
|
|
||||||
signature_b64
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn parse_env(cfg: &mut MyConfig) {
|
fn parse_env(cfg: &mut MyConfig) {
|
||||||
@@ -583,71 +584,45 @@ fn parse_env_netconfig(cfg_lock: &mut MyConfig, chain: &str) -> NetworkParams {
|
|||||||
"testnet4" => &mut cfg_lock.testnet4,
|
"testnet4" => &mut cfg_lock.testnet4,
|
||||||
&_ => &mut cfg_lock.mainnet,
|
&_ => &mut cfg_lock.mainnet,
|
||||||
};
|
};
|
||||||
match env::var(format!("BAL_PUSHER_{}_HOST", chain.to_uppercase())) {
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_HOST", chain.to_uppercase())) {
|
||||||
Ok(value) => {
|
|
||||||
cfg.host = value;
|
cfg.host = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_PORT", chain.to_uppercase()))
|
||||||
}
|
&& let Ok(port_num) = value.parse::<u64>()
|
||||||
match env::var(format!("BAL_PUSHER_{}_PORT", chain.to_uppercase())) {
|
{
|
||||||
Ok(value) => match value.parse::<u64>() {
|
if let Ok(port) = u16::try_from(port_num) {
|
||||||
Ok(value) => match u16::try_from(value) {
|
cfg.port = port;
|
||||||
Ok(port) => cfg.port = port,
|
} else {
|
||||||
Err(e) => {
|
|
||||||
error!(
|
error!(
|
||||||
"Port value {} exceeds u16 range for chain {}: {}",
|
"Port value {} exceeds u16 range for chain {}",
|
||||||
value, chain, e
|
port_num, chain
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
},
|
|
||||||
Err(_) => {}
|
|
||||||
},
|
|
||||||
Err(_) => {}
|
|
||||||
}
|
}
|
||||||
match env::var(format!("BAL_PUSHER_{}_DIR_PATH", chain.to_uppercase())) {
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_DIR_PATH", chain.to_uppercase())) {
|
||||||
Ok(value) => {
|
|
||||||
cfg.dir_path = value;
|
cfg.dir_path = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_DB_FIELD", chain.to_uppercase())) {
|
||||||
}
|
|
||||||
match env::var(format!("BAL_PUSHER_{}_DB_FIELD", chain.to_uppercase())) {
|
|
||||||
Ok(value) => {
|
|
||||||
cfg.db_field = value;
|
cfg.db_field = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_COOKIE_FILE", chain.to_uppercase())) {
|
||||||
}
|
|
||||||
match env::var(format!("BAL_PUSHER_{}_COOKIE_FILE", chain.to_uppercase())) {
|
|
||||||
Ok(value) => {
|
|
||||||
cfg.cookie_file = value;
|
cfg.cookie_file = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_RPC_USER", chain.to_uppercase())) {
|
||||||
}
|
|
||||||
match env::var(format!("BAL_PUSHER_{}_RPC_USER", chain.to_uppercase())) {
|
|
||||||
Ok(value) => {
|
|
||||||
cfg.rpc_user = value;
|
cfg.rpc_user = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_RPC_PASSWORD", chain.to_uppercase())) {
|
||||||
}
|
|
||||||
match env::var(format!("BAL_PUSHER_{}_RPC_PASSWORD", chain.to_uppercase())) {
|
|
||||||
Ok(value) => {
|
|
||||||
cfg.rpc_pass = value;
|
cfg.rpc_pass = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
println!("BAL_PUSHER_{}_ZMQ_HASHBLOCK", chain.to_uppercase());
|
||||||
}
|
if let Ok(value) = env::var(format!("BAL_PUSHER_{}_ZMQ_HASHBLOCK", chain.to_uppercase())) {
|
||||||
println!(
|
|
||||||
"{}",
|
|
||||||
format!("BAL_PUSHER_{}_ZMQ_HASHBLOCK", chain.to_uppercase())
|
|
||||||
);
|
|
||||||
match env::var(format!("BAL_PUSHER_{}_ZMQ_HASHBLOCK", chain.to_uppercase())) {
|
|
||||||
Ok(value) => {
|
|
||||||
println!("value:{}", value);
|
println!("value:{}", value);
|
||||||
cfg.zmq_listener = value;
|
cfg.zmq_listener = value;
|
||||||
}
|
}
|
||||||
Err(_) => {}
|
|
||||||
}
|
|
||||||
cfg.clone()
|
cfg.clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[allow(dead_code)]
|
||||||
fn check_zmq_connection(endpoint: &str) -> bool {
|
fn check_zmq_connection(endpoint: &str) -> bool {
|
||||||
trace!("check zmq connection");
|
trace!("check zmq connection");
|
||||||
let context = Context::new();
|
let context = Context::new();
|
||||||
@@ -665,6 +640,7 @@ fn check_zmq_connection(endpoint: &str) -> bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Add this struct to monitor connection health
|
// Add this struct to monitor connection health
|
||||||
|
#[allow(dead_code)]
|
||||||
struct ConnectionMonitor {
|
struct ConnectionMonitor {
|
||||||
last_message_time: Instant,
|
last_message_time: Instant,
|
||||||
timeout: Duration,
|
timeout: Duration,
|
||||||
@@ -672,6 +648,7 @@ struct ConnectionMonitor {
|
|||||||
max_consecutive_timeouts: u32,
|
max_consecutive_timeouts: u32,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[allow(dead_code)]
|
||||||
impl ConnectionMonitor {
|
impl ConnectionMonitor {
|
||||||
fn new(timeout_secs: u64, max_timeouts: u32) -> Self {
|
fn new(timeout_secs: u64, max_timeouts: u32) -> Self {
|
||||||
Self {
|
Self {
|
||||||
@@ -709,6 +686,7 @@ impl ConnectionMonitor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[allow(dead_code)]
|
||||||
enum ConnectionStatus {
|
enum ConnectionStatus {
|
||||||
Healthy,
|
Healthy,
|
||||||
Warning(Duration),
|
Warning(Duration),
|
||||||
@@ -720,7 +698,6 @@ async fn main() -> std::io::Result<()> {
|
|||||||
env_logger::init();
|
env_logger::init();
|
||||||
let mut cfg = MyConfig::default();
|
let mut cfg = MyConfig::default();
|
||||||
|
|
||||||
let dbfile = env::var("BAL_PUSHER_DB_FILE").unwrap();
|
|
||||||
parse_env(&mut cfg);
|
parse_env(&mut cfg);
|
||||||
let mut args = std::env::args();
|
let mut args = std::env::args();
|
||||||
let _exe_name = args.next().unwrap();
|
let _exe_name = args.next().unwrap();
|
||||||
@@ -755,31 +732,48 @@ async fn main() -> std::io::Result<()> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
match socket.set_subscribe(b"") {
|
match socket.set_subscribe(b"") {
|
||||||
Ok(_) => {}
|
Ok(_) => {
|
||||||
|
info!("ZMQ subscribed to all topics on {}", zmq_address);
|
||||||
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("ZMQ subscribe failed: {}, exiting", e);
|
error!("ZMQ subscribe failed: {}, exiting", e);
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let _ = main_result(&cfg, network_params).await;
|
if let Err(e) = main_result(&cfg, network_params).await {
|
||||||
|
error!("main_result failed on startup: {}", e);
|
||||||
|
}
|
||||||
info!("waiting new blocks..");
|
info!("waiting new blocks..");
|
||||||
let mut last_seq: Vec<u8> = [0; 4].to_vec();
|
|
||||||
let mut counter = 0;
|
|
||||||
let max = 100;
|
|
||||||
socket.set_rcvtimeo(5000).unwrap(); // 5 seconds timeout
|
socket.set_rcvtimeo(5000).unwrap(); // 5 seconds timeout
|
||||||
|
let mut consecutive_timeouts: u32 = 0;
|
||||||
loop {
|
loop {
|
||||||
let message = match socket.recv_multipart(0) {
|
let message = match socket.recv_multipart(0) {
|
||||||
Ok(m) => m,
|
Ok(m) => m,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
consecutive_timeouts += 1;
|
||||||
|
if consecutive_timeouts == 1 {
|
||||||
warn!("ZMQ recv timeout or error: {}, retrying...", e);
|
warn!("ZMQ recv timeout or error: {}, retrying...", e);
|
||||||
|
} else if consecutive_timeouts.is_multiple_of(12) {
|
||||||
|
warn!(
|
||||||
|
"No ZMQ messages for {}s ({} consecutive timeouts), is bitcoind ZMQ active on {}?",
|
||||||
|
consecutive_timeouts * 5,
|
||||||
|
consecutive_timeouts,
|
||||||
|
zmq_address
|
||||||
|
);
|
||||||
|
}
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
if consecutive_timeouts > 0 {
|
||||||
|
info!(
|
||||||
|
"ZMQ connection restored after {} consecutive timeouts",
|
||||||
|
consecutive_timeouts
|
||||||
|
);
|
||||||
|
}
|
||||||
|
consecutive_timeouts = 0;
|
||||||
let topic = message[0].clone();
|
let topic = message[0].clone();
|
||||||
let body = message[1].clone();
|
let body = message[1].clone();
|
||||||
let seq = message[2].clone();
|
|
||||||
last_seq = seq;
|
|
||||||
debug!(
|
debug!(
|
||||||
"ZMQ:GET TOPIC: {}",
|
"ZMQ:GET TOPIC: {}",
|
||||||
String::from_utf8(topic.clone()).expect("invalid topic")
|
String::from_utf8(topic.clone()).expect("invalid topic")
|
||||||
@@ -787,12 +781,15 @@ async fn main() -> std::io::Result<()> {
|
|||||||
trace!("ZMQ:GET BODY: {}", hex::encode(&body));
|
trace!("ZMQ:GET BODY: {}", hex::encode(&body));
|
||||||
if topic == b"hashblock" {
|
if topic == b"hashblock" {
|
||||||
info!("NEW BLOCK: {}", hex::encode(&body));
|
info!("NEW BLOCK: {}", hex::encode(&body));
|
||||||
let _ = main_result(&cfg, network_params).await;
|
if let Err(e) = main_result(&cfg, network_params).await {
|
||||||
|
error!("main_result failed on new block: {}", e);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
thread::sleep(Duration::from_millis(100)); // Sleep for 100ms
|
thread::sleep(Duration::from_millis(100)); // Sleep for 100ms
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
fn seq_to_str(seq: &Vec<u8>) -> String {
|
#[allow(dead_code)]
|
||||||
|
fn seq_to_str(seq: &[u8]) -> String {
|
||||||
if seq.len() == 4 {
|
if seq.len() == 4 {
|
||||||
let mut rdr = Cursor::new(seq);
|
let mut rdr = Cursor::new(seq);
|
||||||
let sequence = rdr
|
let sequence = rdr
|
||||||
|
|||||||
@@ -7,7 +7,6 @@ use chrono::Utc;
|
|||||||
use hex_conservative::FromHex;
|
use hex_conservative::FromHex;
|
||||||
use log::{debug, error, info, trace};
|
use log::{debug, error, info, trace};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use serde_json;
|
|
||||||
use sqlite::State;
|
use sqlite::State;
|
||||||
use sqlite::{Connection, Value};
|
use sqlite::{Connection, Value};
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
@@ -119,6 +118,7 @@ pub struct StatsResponse {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
|
#[allow(dead_code)]
|
||||||
struct ActixConfig {
|
struct ActixConfig {
|
||||||
max_body_size: usize,
|
max_body_size: usize,
|
||||||
timeout_secs: u64,
|
timeout_secs: u64,
|
||||||
@@ -208,7 +208,7 @@ async fn echo_pub_key(data: web::Data<AppState>) -> impl Responder {
|
|||||||
"Failed to read public key file {}: {}",
|
"Failed to read public key file {}: {}",
|
||||||
data.cfg.pub_key_path, e
|
data.cfg.pub_key_path, e
|
||||||
);
|
);
|
||||||
HttpResponse::InternalServerError().body("Failed to read public key file")
|
HttpResponse::InternalServerError().body("error")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -224,13 +224,13 @@ async fn echo_info(
|
|||||||
) -> impl Responder {
|
) -> impl Responder {
|
||||||
let param = path.into_inner();
|
let param = path.into_inner();
|
||||||
if !NETWORKS.contains(¶m.as_str()) {
|
if !NETWORKS.contains(¶m.as_str()) {
|
||||||
return HttpResponse::NotFound().body("Unknown network");
|
return HttpResponse::NotFound().body("error");
|
||||||
}
|
}
|
||||||
info!("echo info!!!{}", param);
|
info!("echo info!!!{}", param);
|
||||||
let netconfig = data.cfg.get_net_config(¶m);
|
let netconfig = data.cfg.get_net_config(¶m);
|
||||||
if !netconfig.enabled {
|
if !netconfig.enabled {
|
||||||
debug!("network disabled {}", param);
|
debug!("network disabled {}", param);
|
||||||
return HttpResponse::BadRequest().body("network disabled");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
let remote_addr = req
|
let remote_addr = req
|
||||||
.headers()
|
.headers()
|
||||||
@@ -257,7 +257,7 @@ async fn echo_info(
|
|||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned in echo_info (lookup phase)");
|
error!("DB mutex poisoned in echo_info (lookup phase)");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
match get_last_used_address_by_ip(
|
match get_last_used_address_by_ip(
|
||||||
@@ -275,10 +275,7 @@ async fn echo_info(
|
|||||||
version: VERSION.to_string(),
|
version: VERSION.to_string(),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
None => {
|
None => get_next_address_index(&db, &netconfig.name, &netconfig.address),
|
||||||
let next = get_next_address_index(&db, &netconfig.name, &netconfig.address);
|
|
||||||
next
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}; // lock released
|
}; // lock released
|
||||||
|
|
||||||
@@ -288,8 +285,7 @@ async fn echo_info(
|
|||||||
Ok(address) => address,
|
Ok(address) => address,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("Failed to derive address from xpub: {}", e);
|
error!("Failed to derive address from xpub: {}", e);
|
||||||
return HttpResponse::BadRequest()
|
return HttpResponse::BadRequest().body("error");
|
||||||
.body(format!("Failed to derive address: {}", e));
|
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -299,7 +295,7 @@ async fn echo_info(
|
|||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned in echo_info (save phase)");
|
error!("DB mutex poisoned in echo_info (save phase)");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
save_new_address(&db, next_idx.0, &derived.0, &derived.1, &remote_addr);
|
save_new_address(&db, next_idx.0, &derived.0, &derived.1, &remote_addr);
|
||||||
@@ -322,30 +318,30 @@ async fn echo_info(
|
|||||||
debug!("echo info reply: {}", json_data);
|
debug!("echo info reply: {}", json_data);
|
||||||
HttpResponse::Ok().json(info)
|
HttpResponse::Ok().json(info)
|
||||||
}
|
}
|
||||||
Err(err) => HttpResponse::InternalServerError().body(format!("error:{}", err)),
|
Err(_err) => HttpResponse::InternalServerError().body("error"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn echo_stats(path: web::Path<String>, data: web::Data<AppState>) -> impl Responder {
|
async fn echo_stats(path: web::Path<String>, data: web::Data<AppState>) -> impl Responder {
|
||||||
let param = path.into_inner();
|
let param = path.into_inner();
|
||||||
if !NETWORKS.contains(¶m.as_str()) {
|
if !NETWORKS.contains(¶m.as_str()) {
|
||||||
return HttpResponse::NotFound().body("Unknown network");
|
return HttpResponse::NotFound().body("error");
|
||||||
}
|
}
|
||||||
info!("echo stats!!! {}", data.cfg.expose_stats);
|
info!("echo stats!!! {}", data.cfg.expose_stats);
|
||||||
let netconfig = data.cfg.get_net_config(¶m);
|
let netconfig = data.cfg.get_net_config(¶m);
|
||||||
if !netconfig.enabled {
|
if !netconfig.enabled {
|
||||||
debug!("network disabled {}", param);
|
debug!("network disabled {}", param);
|
||||||
return HttpResponse::BadRequest().body("network disabled");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
if !data.cfg.expose_stats {
|
if !data.cfg.expose_stats {
|
||||||
return HttpResponse::Forbidden().body("Stats not exposed");
|
return HttpResponse::Forbidden().body("error");
|
||||||
}
|
}
|
||||||
let mut stats: Vec<StatsResponse> = vec![];
|
let mut stats: Vec<StatsResponse> = vec![];
|
||||||
let db = match data.db.lock() {
|
let db = match data.db.lock() {
|
||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned in echo_stats");
|
error!("DB mutex poisoned in echo_stats");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let mut stmt = match db.prepare(
|
let mut stmt = match db.prepare(
|
||||||
@@ -354,12 +350,12 @@ async fn echo_stats(path: web::Path<String>, data: web::Data<AppState>) -> impl
|
|||||||
Ok(s) => s,
|
Ok(s) => s,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("Failed to prepare stats query: {}", e);
|
error!("Failed to prepare stats query: {}", e);
|
||||||
return HttpResponse::InternalServerError().body("Database error");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if let Err(e) = stmt.bind((1, Value::String(netconfig.name.clone()))) {
|
if let Err(e) = stmt.bind((1, Value::String(netconfig.name.clone()))) {
|
||||||
error!("Failed to bind chain in stats query: {}", e);
|
error!("Failed to bind chain in stats query: {}", e);
|
||||||
return HttpResponse::InternalServerError().body("Database error");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
while let Ok(State::Row) = stmt.next() {
|
while let Ok(State::Row) = stmt.next() {
|
||||||
let report_date = stmt.read("report_date").unwrap_or("0".to_string());
|
let report_date = stmt.read("report_date").unwrap_or("0".to_string());
|
||||||
@@ -422,7 +418,7 @@ async fn echo_stats(path: web::Path<String>, data: web::Data<AppState>) -> impl
|
|||||||
debug!("echo info reply: {}", json_data);
|
debug!("echo info reply: {}", json_data);
|
||||||
HttpResponse::Ok().json(stats)
|
HttpResponse::Ok().json(stats)
|
||||||
}
|
}
|
||||||
Err(err) => HttpResponse::InternalServerError().body(format!("error:{}", err)),
|
Err(_err) => HttpResponse::InternalServerError().body("error"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -431,33 +427,33 @@ async fn echo_search(body: Bytes, data: web::Data<AppState>) -> impl Responder {
|
|||||||
let strbody = match std::str::from_utf8(&body) {
|
let strbody = match std::str::from_utf8(&body) {
|
||||||
Ok(s) => s,
|
Ok(s) => s,
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
return HttpResponse::BadRequest().body("Invalid UTF-8 body");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
info!("{}", strbody);
|
info!("{}", strbody);
|
||||||
|
|
||||||
if strbody.is_empty() || strbody.len() != 64 || !strbody.chars().all(|c| c.is_ascii_hexdigit())
|
if strbody.is_empty() || strbody.len() != 64 || !strbody.chars().all(|c| c.is_ascii_hexdigit())
|
||||||
{
|
{
|
||||||
return HttpResponse::BadRequest().body("Invalid txid");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
|
|
||||||
let db = match data.db.lock() {
|
let db = match data.db.lock() {
|
||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned in echo_search");
|
error!("DB mutex poisoned in echo_search");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let mut statement = match db.prepare("SELECT * FROM tbl_tx WHERE txid = ? LIMIT 1") {
|
let mut statement = match db.prepare("SELECT * FROM tbl_tx WHERE txid = ? LIMIT 1") {
|
||||||
Ok(s) => s,
|
Ok(s) => s,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("Failed to prepare statement: {}", e);
|
error!("Failed to prepare statement: {}", e);
|
||||||
return HttpResponse::InternalServerError().body("Database error");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if let Err(e) = statement.bind((1, strbody)) {
|
if let Err(e) = statement.bind((1, strbody)) {
|
||||||
error!("Failed to bind parameter: {}", e);
|
error!("Failed to bind parameter: {}", e);
|
||||||
return HttpResponse::InternalServerError().body("Database error");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
|
|
||||||
if let Ok(State::Row) = statement.next() {
|
if let Ok(State::Row) = statement.next() {
|
||||||
@@ -503,11 +499,14 @@ async fn echo_search(body: Bytes, data: web::Data<AppState>) -> impl Responder {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
match serde_json::to_string(&response_data) {
|
match serde_json::to_string(&response_data) {
|
||||||
Ok(json_data) => HttpResponse::Ok().json(json_data),
|
Ok(json_data) => {
|
||||||
Err(_) => HttpResponse::BadRequest().body("Bad data received"),
|
debug!("echo search reply: {}", json_data);
|
||||||
|
HttpResponse::Ok().json(&response_data)
|
||||||
|
}
|
||||||
|
Err(_) => HttpResponse::BadRequest().body("error"),
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
HttpResponse::BadRequest().body("Bad data received")
|
HttpResponse::BadRequest().body("error")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -524,13 +523,13 @@ struct ParsedTx {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Parse all transactions from the request body **without** needing the DB lock.
|
/// Parse all transactions from the request body **without** needing the DB lock.
|
||||||
/// Returns `Ok(parsed_txs)` if at least one tx was valid, or `Err(HttpResponse)` for early failure.
|
/// Skips transactions that don't have a valid willexecutor output.
|
||||||
fn parse_request_transactions(
|
fn parse_request_transactions(
|
||||||
strbody: &str,
|
strbody: &str,
|
||||||
_req_time: i64,
|
_req_time: i64,
|
||||||
netconfig: &NetConfig,
|
netconfig: &NetConfig,
|
||||||
known_addresses: &HashSet<String>,
|
known_addresses: &HashSet<String>,
|
||||||
) -> Result<Vec<(ParsedTx, String, u64)>, HttpResponse> {
|
) -> Vec<(ParsedTx, String, u64)> {
|
||||||
let mut result: Vec<(ParsedTx, String, u64)> = Vec::new();
|
let mut result: Vec<(ParsedTx, String, u64)> = Vec::new();
|
||||||
let mut union_tx = true;
|
let mut union_tx = true;
|
||||||
|
|
||||||
@@ -613,8 +612,8 @@ fn parse_request_transactions(
|
|||||||
}
|
}
|
||||||
|
|
||||||
if !found {
|
if !found {
|
||||||
error!("willexecutor output not found for tx {}", txid);
|
trace!("willexecutor output not found for tx {}, skipping", txid);
|
||||||
return Err(HttpResponse::BadRequest().body("Bad data received"));
|
continue;
|
||||||
}
|
}
|
||||||
if !union_tx {
|
if !union_tx {
|
||||||
// This is only used for SQL building later; we track it in the caller
|
// This is only used for SQL building later; we track it in the caller
|
||||||
@@ -636,7 +635,7 @@ fn parse_request_transactions(
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(result)
|
result
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn echo_push(
|
async fn echo_push(
|
||||||
@@ -648,24 +647,24 @@ async fn echo_push(
|
|||||||
let strbody = match std::str::from_utf8(&body) {
|
let strbody = match std::str::from_utf8(&body) {
|
||||||
Ok(s) => s,
|
Ok(s) => s,
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
return HttpResponse::BadRequest().body("Invalid UTF-8 body");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
let param = path.into_inner();
|
let param = path.into_inner();
|
||||||
if !NETWORKS.contains(¶m.as_str()) {
|
if !NETWORKS.contains(¶m.as_str()) {
|
||||||
return HttpResponse::NotFound().body("Unknown network");
|
return HttpResponse::NotFound().body("error");
|
||||||
}
|
}
|
||||||
let netconfig = data.cfg.get_net_config(¶m);
|
let netconfig = data.cfg.get_net_config(¶m);
|
||||||
if !netconfig.enabled {
|
if !netconfig.enabled {
|
||||||
trace!("network not enabled {}", &netconfig.name);
|
trace!("network not enabled {}", &netconfig.name);
|
||||||
return HttpResponse::BadRequest().body("Network not enabled");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
let req_time = match Utc::now().timestamp_nanos_opt() {
|
let req_time = match Utc::now().timestamp_nanos_opt() {
|
||||||
Some(t) => t,
|
Some(t) => t,
|
||||||
None => {
|
None => {
|
||||||
error!("Invalid timestamp");
|
error!("Invalid timestamp");
|
||||||
return HttpResponse::BadRequest().body("Invalid timestamp");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -675,7 +674,7 @@ async fn echo_push(
|
|||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned acquiring addresses in echo_push");
|
error!("DB mutex poisoned acquiring addresses in echo_push");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if netconfig.xpub {
|
if netconfig.xpub {
|
||||||
@@ -683,7 +682,7 @@ async fn echo_push(
|
|||||||
Ok(addrs) => addrs,
|
Ok(addrs) => addrs,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("Failed to load addresses from xpub: {}", e);
|
error!("Failed to load addresses from xpub: {}", e);
|
||||||
return HttpResponse::InternalServerError().body("Database error");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
@@ -692,12 +691,9 @@ async fn echo_push(
|
|||||||
}; // lock released here
|
}; // lock released here
|
||||||
|
|
||||||
// Parse all transactions (CPU-bound, no DB needed)
|
// Parse all transactions (CPU-bound, no DB needed)
|
||||||
let parsed = match parse_request_transactions(strbody, req_time, netconfig, &known_addresses) {
|
let parsed = parse_request_transactions(strbody, req_time, netconfig, &known_addresses);
|
||||||
Ok(v) => v,
|
|
||||||
Err(resp) => return resp,
|
|
||||||
};
|
|
||||||
if parsed.is_empty() {
|
if parsed.is_empty() {
|
||||||
return HttpResponse::Ok().body("thx");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
|
|
||||||
let all_txids: Vec<String> = parsed.iter().map(|(p, _, _)| p.txid.clone()).collect();
|
let all_txids: Vec<String> = parsed.iter().map(|(p, _, _)| p.txid.clone()).collect();
|
||||||
@@ -708,14 +704,14 @@ async fn echo_push(
|
|||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned in echo_push duplicate check");
|
error!("DB mutex poisoned in echo_push duplicate check");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
match check_duplicate_txids(&db, &all_txids) {
|
match check_duplicate_txids(&db, &all_txids) {
|
||||||
Ok(dups) => dups,
|
Ok(dups) => dups,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("Duplicate check failed: {}", e);
|
error!("Duplicate check failed: {}", e);
|
||||||
return HttpResponse::InternalServerError().body("Database error");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}; // lock released here
|
}; // lock released here
|
||||||
@@ -731,7 +727,7 @@ async fn echo_push(
|
|||||||
Ok(g) => g,
|
Ok(g) => g,
|
||||||
Err(_p) => {
|
Err(_p) => {
|
||||||
error!("DB mutex poisoned in echo_push insert phase");
|
error!("DB mutex poisoned in echo_push insert phase");
|
||||||
return HttpResponse::InternalServerError().body("DB mutex poisoned");
|
return HttpResponse::InternalServerError().body("error");
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -818,7 +814,7 @@ async fn echo_push(
|
|||||||
|
|
||||||
if let Err(err) = execute_insert(&db, sqltxs, ptx, sqlinps, pinps, sqlouts, pouts) {
|
if let Err(err) = execute_insert(&db, sqltxs, ptx, sqlinps, pinps, sqlouts, pouts) {
|
||||||
error!("execute_insert failed: {}", err);
|
error!("execute_insert failed: {}", err);
|
||||||
return HttpResponse::BadRequest().body("Bad data received");
|
return HttpResponse::BadRequest().body("error");
|
||||||
}
|
}
|
||||||
} // lock released
|
} // lock released
|
||||||
|
|
||||||
@@ -900,7 +896,7 @@ async fn main() -> std::io::Result<()> {
|
|||||||
let db = match open_db(&cfg.db_file) {
|
let db = match open_db(&cfg.db_file) {
|
||||||
Ok(c) => c,
|
Ok(c) => c,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
return Err(std::io::Error::new(std::io::ErrorKind::Other, e));
|
return Err(std::io::Error::other(e));
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@@ -178,8 +178,8 @@ pub fn create_database(db: &Connection) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
*/
|
*/
|
||||||
pub fn insert_xpub(db: &Connection, network: &String, xpub: &String) {
|
pub fn insert_xpub(db: &Connection, network: &str, xpub: &str) {
|
||||||
if xpub != "" {
|
if !xpub.is_empty() {
|
||||||
trace!("going to insert: {} xpub:{}", network, xpub);
|
trace!("going to insert: {} xpub:{}", network, xpub);
|
||||||
let mut stmt = match db.prepare("INSERT INTO tbl_xpub(network,xpub) VALUES(?, ?);") {
|
let mut stmt = match db.prepare("INSERT INTO tbl_xpub(network,xpub) VALUES(?, ?);") {
|
||||||
Ok(s) => s,
|
Ok(s) => s,
|
||||||
|
|||||||
32
src/xpub.rs
32
src/xpub.rs
@@ -11,6 +11,7 @@ use sha2::{Digest, Sha256};
|
|||||||
use std::str::FromStr;
|
use std::str::FromStr;
|
||||||
|
|
||||||
// Mainnet (BIP44/BIP49/BIP84)
|
// Mainnet (BIP44/BIP49/BIP84)
|
||||||
|
#[allow(dead_code)]
|
||||||
enum BS58Prefix {
|
enum BS58Prefix {
|
||||||
Xpub,
|
Xpub,
|
||||||
Ypub,
|
Ypub,
|
||||||
@@ -102,44 +103,29 @@ fn calc_checksum(desc: &str) -> Result<String, String> {
|
|||||||
Ok(checksum)
|
Ok(checksum)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn get_bitcoincore_descriptor(xpub: &String) -> String {
|
pub fn get_bitcoincore_descriptor(xpub: &str) -> String {
|
||||||
let fingerprint = match calculate_fingerprint(xpub) {
|
let fingerprint = match calculate_fingerprint(xpub) {
|
||||||
Ok(f) => f,
|
Ok(f) => f,
|
||||||
Err(_) => return String::new(), // Invalid xpub, return empty descriptor
|
Err(_) => return String::new(), // Invalid xpub, return empty descriptor
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut bip = 84;
|
|
||||||
let cpub = xpub.to_string();
|
|
||||||
match &xpub[0..4] {
|
|
||||||
"vpub" => {
|
|
||||||
bip = 84;
|
|
||||||
}
|
|
||||||
"zpub" => {
|
|
||||||
bip = 84;
|
|
||||||
}
|
|
||||||
&_ => {
|
|
||||||
bip = 84;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
let xpub_converted = match convert_xpub(xpub) {
|
let xpub_converted = match convert_xpub(xpub) {
|
||||||
Ok(c) => c,
|
Ok(c) => c,
|
||||||
Err(_) => return String::new(), // Invalid xpub, return empty descriptor
|
Err(_) => return String::new(), // Invalid xpub, return empty descriptor
|
||||||
};
|
};
|
||||||
let descriptor = format!("wpkh([{}/84h/0h/0h]{}/0/*)", fingerprint, xpub_converted);
|
let descriptor = format!("wpkh([{}/84h/0h/0h]{}/0/*)", fingerprint, xpub_converted);
|
||||||
let descriptor = match calc_checksum(&descriptor) {
|
match calc_checksum(&descriptor) {
|
||||||
Ok(checksum) => {
|
Ok(checksum) => {
|
||||||
let clean_descriptor = descriptor.split('#').next().unwrap_or(&descriptor);
|
let clean_descriptor = descriptor.split('#').next().unwrap_or(&descriptor);
|
||||||
format!("{}#{}", clean_descriptor, checksum)
|
format!("{}#{}", clean_descriptor, checksum)
|
||||||
}
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
eprintln!("Error: {}", err);
|
eprintln!("Error: {}", err);
|
||||||
"".to_string()
|
String::new()
|
||||||
}
|
}
|
||||||
};
|
|
||||||
descriptor
|
|
||||||
//format!("{}#{}",descriptor,checksum)
|
|
||||||
}
|
}
|
||||||
fn convert_xpub(xpub: &String) -> Result<String, String> {
|
}
|
||||||
|
fn convert_xpub(xpub: &str) -> Result<String, String> {
|
||||||
if xpub.len() >= 4 && (&xpub[0..4] == "xpub" || &xpub[0..4] == "ypub" || &xpub[0..4] == "zpub")
|
if xpub.len() >= 4 && (&xpub[0..4] == "xpub" || &xpub[0..4] == "ypub" || &xpub[0..4] == "zpub")
|
||||||
{
|
{
|
||||||
convert_to(xpub, BS58Prefix::Xpub)
|
convert_to(xpub, BS58Prefix::Xpub)
|
||||||
@@ -165,7 +151,7 @@ fn base58check_decode(s: &str) -> Result<Vec<u8>, String> {
|
|||||||
return Err("Data troppo corta".to_string());
|
return Err("Data troppo corta".to_string());
|
||||||
}
|
}
|
||||||
let (payload, checksum) = data.split_at(data.len() - 4);
|
let (payload, checksum) = data.split_at(data.len() - 4);
|
||||||
let hash = Sha256::digest(&Sha256::digest(payload));
|
let hash = Sha256::digest(Sha256::digest(payload));
|
||||||
if hash[0..4] != checksum[..] {
|
if hash[0..4] != checksum[..] {
|
||||||
return Err("Checksum invalido".to_string());
|
return Err("Checksum invalido".to_string());
|
||||||
}
|
}
|
||||||
@@ -173,7 +159,7 @@ fn base58check_decode(s: &str) -> Result<Vec<u8>, String> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn base58check_encode(data: &[u8]) -> String {
|
fn base58check_encode(data: &[u8]) -> String {
|
||||||
let checksum = &Sha256::digest(&Sha256::digest(data))[0..4];
|
let checksum = &Sha256::digest(Sha256::digest(data))[0..4];
|
||||||
let full = [data, checksum].concat();
|
let full = [data, checksum].concat();
|
||||||
bs58::encode(full).into_string()
|
bs58::encode(full).into_string()
|
||||||
}
|
}
|
||||||
@@ -205,7 +191,7 @@ pub fn new_address_from_xpub(
|
|||||||
) -> Result<(String, String), Box<dyn std::error::Error>> {
|
) -> Result<(String, String), Box<dyn std::error::Error>> {
|
||||||
let xpub = Xpub::from_str(&convert_to(zpub, BS58Prefix::Xpub)?)?;
|
let xpub = Xpub::from_str(&convert_to(zpub, BS58Prefix::Xpub)?)?;
|
||||||
let path = format!("m/0/{}", index);
|
let path = format!("m/0/{}", index);
|
||||||
let derivation_path = DerivationPath::from_str(&path.as_str())?;
|
let derivation_path = DerivationPath::from_str(path.as_str())?;
|
||||||
let secp = Secp256k1::new();
|
let secp = Secp256k1::new();
|
||||||
let derived_xpub = xpub.derive_pub(&secp, &derivation_path)?;
|
let derived_xpub = xpub.derive_pub(&secp, &derivation_path)?;
|
||||||
let public_key = derived_xpub.public_key;
|
let public_key = derived_xpub.public_key;
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
use bal_server::db::open_db;
|
use bal_server::db::open_db;
|
||||||
use sqlite::State;
|
use sqlite::State;
|
||||||
use std::fs;
|
use std::fs;
|
||||||
use std::path::Path;
|
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_open_db_blocks_traversal() {
|
fn test_open_db_blocks_traversal() {
|
||||||
@@ -74,7 +73,7 @@ fn test_open_db_rejects_symlink() {
|
|||||||
let _ = fs::remove_file(real);
|
let _ = fs::remove_file(real);
|
||||||
let _ = fs::remove_file(link);
|
let _ = fs::remove_file(link);
|
||||||
fs::File::create(real).unwrap();
|
fs::File::create(real).unwrap();
|
||||||
fs::soft_link(real, link).unwrap();
|
std::os::unix::fs::symlink(real, link).unwrap();
|
||||||
|
|
||||||
let res = open_db(link);
|
let res = open_db(link);
|
||||||
assert!(res.is_err(), "Symlink DB path should be rejected");
|
assert!(res.is_err(), "Symlink DB path should be rejected");
|
||||||
|
|||||||
@@ -36,11 +36,8 @@ fn test_db_null_unwrap_or() {
|
|||||||
|
|
||||||
let mut found_value = None;
|
let mut found_value = None;
|
||||||
let _ = db.iterate("SELECT * FROM test_stats;", |pairs| {
|
let _ = db.iterate("SELECT * FROM test_stats;", |pairs| {
|
||||||
let row: HashMap<_, _> = pairs
|
let row: HashMap<_, _> = pairs.iter().map(|(k, v)| (k.to_string(), *v)).collect();
|
||||||
.into_iter()
|
let totals = row["totals"].unwrap_or("0").to_string();
|
||||||
.map(|(k, v)| (k.to_string(), v.map(|s| s)))
|
|
||||||
.collect();
|
|
||||||
let totals = row["totals"].clone().unwrap_or("0").to_string();
|
|
||||||
found_value = Some(totals);
|
found_value = Some(totals);
|
||||||
true
|
true
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
use std::fs;
|
use std::fs;
|
||||||
use std::path::Path;
|
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_gitignore_protection_env() {
|
fn test_gitignore_protection_env() {
|
||||||
@@ -23,30 +22,20 @@ fn test_gitignore_protection_env() {
|
|||||||
];
|
];
|
||||||
|
|
||||||
for pattern in required_patterns {
|
for pattern in required_patterns {
|
||||||
let has_exact = gitignore.contains(&pattern);
|
let has_wildcard = gitignore.contains("*.env.local") || gitignore.contains(".env.local");
|
||||||
let has_wildcard = gitignore.contains(&format!("*.env.local"))
|
|
||||||
|| gitignore.contains(&format!(".env.local"));
|
|
||||||
let has_env = gitignore.contains("*.env") || gitignore.contains(".env");
|
let has_env = gitignore.contains("*.env") || gitignore.contains(".env");
|
||||||
|
|
||||||
// For .env.local, either .env.local or *.env.local is acceptable
|
|
||||||
let is_env_local = pattern == "*.env.local" || pattern == ".env.local";
|
let is_env_local = pattern == "*.env.local" || pattern == ".env.local";
|
||||||
if is_env_local {
|
if is_env_local {
|
||||||
assert!(
|
assert!(
|
||||||
has_wildcard,
|
has_wildcard,
|
||||||
".gitignore must contain pattern '*.env.local' or '.env.local' to protect secrets",
|
".gitignore must contain pattern '*.env.local' or '.env.local' to protect secrets",
|
||||||
);
|
);
|
||||||
} else if pattern == ".env.production" || pattern == ".env.secret" {
|
} else if pattern == ".env.production"
|
||||||
assert!(
|
|| pattern == ".env.secret"
|
||||||
gitignore.contains(pattern),
|
|| pattern == "*.env"
|
||||||
".gitignore must contain pattern '{}' to protect secrets",
|
|| pattern == ".env"
|
||||||
pattern
|
{
|
||||||
);
|
|
||||||
} else if pattern == "*.env" {
|
|
||||||
assert!(
|
|
||||||
has_env,
|
|
||||||
".gitignore must contain pattern '*.env' or '.env' to protect secrets",
|
|
||||||
);
|
|
||||||
} else if pattern == ".env" {
|
|
||||||
assert!(
|
assert!(
|
||||||
has_env,
|
has_env,
|
||||||
".gitignore must contain pattern '*.env' or '.env' to protect secrets",
|
".gitignore must contain pattern '*.env' or '.env' to protect secrets",
|
||||||
@@ -65,7 +54,6 @@ fn test_gitignore_protection_env() {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_no_private_key_in_git() {
|
fn test_no_private_key_in_git() {
|
||||||
// Check that .gitignore includes private_key.pem
|
|
||||||
let gitignore = match fs::read_to_string(".gitignore") {
|
let gitignore = match fs::read_to_string(".gitignore") {
|
||||||
Ok(c) => c,
|
Ok(c) => c,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
@@ -88,20 +76,17 @@ fn test_no_private_key_in_git() {
|
|||||||
".gitignore must block chiave_privata.key"
|
".gitignore must block chiave_privata.key"
|
||||||
);
|
);
|
||||||
|
|
||||||
// Check that no private key files are tracked by git
|
|
||||||
let output = std::process::Command::new("git")
|
let output = std::process::Command::new("git")
|
||||||
.args(&["ls-files", "*.pem", "*.key"])
|
.args(["ls-files", "*.pem", "*.key"])
|
||||||
.output()
|
.output()
|
||||||
.expect("Failed to run git ls-files");
|
.expect("Failed to run git ls-files");
|
||||||
|
|
||||||
let tracked_keys = String::from_utf8(output.stdout).unwrap();
|
let tracked_keys = String::from_utf8(output.stdout).unwrap();
|
||||||
let tracked_keys: Vec<&str> = tracked_keys.lines().collect();
|
let tracked_keys: Vec<&str> = tracked_keys.lines().collect();
|
||||||
|
|
||||||
// Only non-empty entries and only public_key.pem should be tracked
|
|
||||||
for tracked in tracked_keys.iter().filter(|s| !s.is_empty()) {
|
for tracked in tracked_keys.iter().filter(|s| !s.is_empty()) {
|
||||||
if !tracked.contains("public_key.pem") {
|
if !tracked.contains("public_key.pem") {
|
||||||
assert!(
|
panic!(
|
||||||
false,
|
|
||||||
"Private key file is tracked by git: {}. Remove it with git rm --cached",
|
"Private key file is tracked by git: {}. Remove it with git rm --cached",
|
||||||
tracked
|
tracked
|
||||||
);
|
);
|
||||||
@@ -113,38 +98,34 @@ fn test_no_private_key_in_git() {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_no_token_in_source_files() {
|
fn test_no_token_in_source_files() {
|
||||||
// Scan source files for hardcoded tokens
|
|
||||||
let mut found_issues = Vec::new();
|
let mut found_issues = Vec::new();
|
||||||
|
|
||||||
// Scan .sh files for hardcoded 40-char hex strings
|
|
||||||
for entry in fs::read_dir(".").unwrap().filter_map(|e| e.ok()) {
|
for entry in fs::read_dir(".").unwrap().filter_map(|e| e.ok()) {
|
||||||
let path = entry.path();
|
let path = entry.path();
|
||||||
if !path.is_file() {
|
if !path.is_file() {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
if let Some(ext) = path.extension() {
|
if let Some(ext) = path.extension()
|
||||||
if ext == "sh" {
|
&& ext == "sh"
|
||||||
|
{
|
||||||
let content = fs::read_to_string(&path).unwrap();
|
let content = fs::read_to_string(&path).unwrap();
|
||||||
for (line_num, line) in content.lines().enumerate() {
|
for (line_num, line) in content.lines().enumerate() {
|
||||||
// Skip comments and example/template files
|
if line.trim().starts_with('#')
|
||||||
if line.trim().starts_with("#")
|
|
||||||
|| line.to_lowercase().contains("example")
|
|| line.to_lowercase().contains("example")
|
||||||
|| line.to_lowercase().contains("template")
|
|| line.to_lowercase().contains("template")
|
||||||
{
|
{
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
// Check for 40-64 hex chars that could be API tokens (not in .env.example comments)
|
|
||||||
if line.trim().len() >= 40 {
|
if line.trim().len() >= 40 {
|
||||||
let hex_chars = line
|
let hex_chars: Vec<_> = line
|
||||||
.trim()
|
.trim()
|
||||||
.chars()
|
.chars()
|
||||||
.filter(|c| c.is_ascii_hexdigit())
|
.filter(|c| c.is_ascii_hexdigit())
|
||||||
.collect::<Vec<_>>();
|
.collect();
|
||||||
if hex_chars.len() >= 40 && hex_chars.len() <= 64 {
|
if (40..=64).contains(&hex_chars.len())
|
||||||
// Check if it looks like it's part of a TOKEN assignment
|
&& (line.to_lowercase().contains("token")
|
||||||
if line.to_lowercase().contains("token")
|
|
||||||
|| line.to_lowercase().contains("api")
|
|| line.to_lowercase().contains("api")
|
||||||
|| line.to_lowercase().contains("secret")
|
|| line.to_lowercase().contains("secret"))
|
||||||
{
|
{
|
||||||
found_issues.push(format!(
|
found_issues.push(format!(
|
||||||
"Potential hardcoded token in {}: line {}: {}",
|
"Potential hardcoded token in {}: line {}: {}",
|
||||||
@@ -157,16 +138,13 @@ fn test_no_token_in_source_files() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if !found_issues.is_empty() {
|
if !found_issues.is_empty() {
|
||||||
println!("FAIL: Found potential hardcoded tokens:");
|
println!("FAIL: Found potential hardcoded tokens:");
|
||||||
for issue in &found_issues {
|
for issue in &found_issues {
|
||||||
println!(" {}", issue);
|
println!(" {}", issue);
|
||||||
}
|
}
|
||||||
assert!(
|
panic!(
|
||||||
false,
|
|
||||||
"Found potential hardcoded tokens in shell scripts: {:?}",
|
"Found potential hardcoded tokens in shell scripts: {:?}",
|
||||||
found_issues
|
found_issues
|
||||||
);
|
);
|
||||||
|
|||||||
Reference in New Issue
Block a user