4nk/sdk_relay
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: 4nk:add_idnot_service
Branches Tags
4nk:main
4nk:dev
4nk:docker-support-v2
4nk:docker-support
4nk:docker-fixes
4nk:blindbit
4nk:cicd
4nk:add_idnot_service
4nk:v0.1.1
..
compare: 4nk:main
Branches Tags
4nk:dev
4nk:docker-support-v2
4nk:docker-support
4nk:docker-fixes
4nk:blindbit
4nk:cicd
4nk:add_idnot_service
4nk:main
4nk:v0.1.1

No commits in common. "add_idnot_service" and "main" have entirely different histories.
add_idnot_ ... main

13 changed files with 0 additions and 6418 deletions
Show Stats Expand all files Collapse all files

6
.conf.model
View File

@ -1,6 +0,0 @@
core_url=""
ws_url=""
wallet_name="default"
network="signet"
electrum_url="tcp://localhost:60601"
zmq_url=""

2
.gitignore vendored
View File

@ -1,2 +0,0 @@
/target
.conf

3518
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

26
Cargo.toml
View File

@ -1,26 +0,0 @@
[package]
name = "sdk_relay"
version = "0.1.0"
edition = "2021"
[dependencies]
anyhow = "1.0"
bitcoincore-rpc = { version = "0.18" }
electrum-client = { git = "https://github.com/cygnet3/rust-electrum-client", branch = "sp_tweaks" }
env_logger = "0.9"
futures-util = { version = "0.3.28", default-features = false, features = ["sink", "std"] }
hex = "0.4.3"
jwt-simple = { version = "0.12.12", default-features = false, features = ["pure-rust"] }
log = "0.4.20"
reqwest = { version = "0.12.15", features = ["json"] }
sdk_common = { git = "https://git.4nkweb.com/4nk/sdk_common.git", branch = "dev" }
serde = { version = "1.0.193", features = ["derive"]}
serde_json = "1.0"
serde_with = "3.6.0"
tokio = { version = "1.0.0", features = ["io-util", "rt-multi-thread", "macros", "sync"] }
tokio-stream = "0.1"
tokio-tungstenite = "0.21.0"
zeromq = "0.4.1"
[dev-dependencies]
mockall = "0.13.0"

764
src/commit.rs
View File

@ -1,764 +0,0 @@
use std::{
collections::HashMap,
sync::{Mutex, MutexGuard, OnceLock},
};
use anyhow::{Error, Result};
use bitcoincore_rpc::bitcoin::hex::DisplayHex;
use sdk_common::network::{AnkFlag, CommitMessage, HandshakeMessage};
use sdk_common::process::{lock_processes, Process, ProcessState};
use sdk_common::serialization::{OutPointMemberMap, OutPointProcessMap};
use sdk_common::silentpayments::create_transaction;
use sdk_common::sp_client::bitcoin::{Amount, OutPoint};
use sdk_common::sp_client::{FeeRate, Recipient};
use sdk_common::{
pcd::Member,
silentpayments::sign_transaction,
sp_client::{silentpayments::SilentPaymentAddress, RecipientAddress},
};
use crate::message::{broadcast_message, BroadcastType};
use crate::{lock_freezed_utxos, MutexExt, DAEMON, STORAGE, WALLET};
pub(crate) fn handle_commit_request(commit_msg: CommitMessage) -> Result<OutPoint> {
let mut processes = lock_processes()?;
if let Some(process) = processes.get_mut(&commit_msg.process_id) {
handle_existing_commitment(process, &commit_msg)?;
} else {
let new_process = handle_new_process(&commit_msg)?;
// Cache the process
processes.insert(commit_msg.process_id, new_process);
}
// Dump to disk
dump_cached_processes(processes.clone())?;
// Add to frozen UTXOs
lock_freezed_utxos()?.insert(commit_msg.process_id);
// Update processes with the
// Send an update to all connected client
let our_sp_address = WALLET
.get()
.ok_or(Error::msg("Wallet not initialized"))?
.lock_anyhow()?
.get_sp_client()
.get_receiving_address();
let mut new_process_map = HashMap::new();
let new_process = processes.get(&commit_msg.process_id).unwrap().clone();
new_process_map.insert(commit_msg.process_id, new_process);
let init_msg = HandshakeMessage::new(
our_sp_address.to_string(),
OutPointMemberMap(HashMap::new()),
OutPointProcessMap(new_process_map),
);
if let Err(e) = broadcast_message(
AnkFlag::Handshake,
format!("{}", init_msg.to_string()),
BroadcastType::ToAll,
) {
log::error!("Failed to send handshake message: {}", e);
}
Ok(commit_msg.process_id)
}
fn handle_new_process(commit_msg: &CommitMessage) -> Result<Process> {
let pcd_commitment = &commit_msg.pcd_commitment;
let merkle_root_bin = pcd_commitment.create_merkle_tree()?.root().unwrap();
if let Ok(pairing_process_id) = handle_member_list(&commit_msg) {
dump_cached_members()?;
// Send a handshake message to every connected client
if let Some(new_member) = lock_members().unwrap().get(&pairing_process_id) {
let our_sp_address = WALLET
.get()
.ok_or(Error::msg("Wallet not initialized"))?
.lock_anyhow()?
.get_sp_client()
.get_receiving_address();
let mut new_member_map = HashMap::new();
new_member_map.insert(pairing_process_id, new_member.clone());
let init_msg = HandshakeMessage::new(
our_sp_address.into(),
OutPointMemberMap(new_member_map),
OutPointProcessMap(HashMap::new()),
);
if let Err(e) = broadcast_message(
AnkFlag::Handshake,
format!("{}", init_msg.to_string()),
BroadcastType::ToAll,
) {
log::error!("Failed to send handshake message: {}", e);
}
} else {
log::error!(
"Failed to find new member with process id {}",
pairing_process_id
);
}
}
let mut new_process = Process::new(commit_msg.process_id);
let init_state = ProcessState {
commited_in: commit_msg.process_id,
roles: commit_msg.roles.clone(),
pcd_commitment: commit_msg.pcd_commitment.clone(),
state_id: merkle_root_bin,
public_data: commit_msg.public_data.clone(),
..Default::default()
};
new_process.insert_concurrent_state(init_state)?;
Ok(new_process)
}
pub static MEMBERLIST: OnceLock<Mutex<HashMap<OutPoint, Member>>> = OnceLock::new();
pub fn lock_members() -> Result<MutexGuard<'static, HashMap<OutPoint, Member>>, anyhow::Error> {
MEMBERLIST
.get_or_init(|| Mutex::new(HashMap::new()))
.lock_anyhow()
}
fn handle_member_list(commit_msg: &CommitMessage) -> Result<OutPoint> {
//Check if there is one role with one member
if commit_msg.roles.len() != 1 {
return Err(Error::msg("Process is not a pairing process"));
}
if let Some(pairing_role) = commit_msg.roles.get("pairing") {
if !pairing_role.members.is_empty() {
return Err(Error::msg("Process is not a pairing process"));
}
} else {
return Err(Error::msg("Process is not a pairing process"));
}
if let Some(paired_addresses) = commit_msg.public_data.get("pairedAddresses") {
let paired_addresses: Vec<SilentPaymentAddress> =
sdk_common::serialization::ciborium_deserialize(paired_addresses)?;
let mut memberlist = lock_members()?;
memberlist.insert(commit_msg.process_id, Member::new(paired_addresses));
return Ok(commit_msg.process_id);
}
Err(Error::msg("Process is not a pairing process"))
}
fn handle_existing_commitment(
process_to_udpate: &mut Process,
commit_msg: &CommitMessage,
) -> Result<()> {
let process_id = process_to_udpate.get_process_id()?;
match register_new_state(process_to_udpate, &commit_msg) {
Ok(new_state_id) => log::debug!(
"Registering new state for process {} with state id {}",
process_id,
new_state_id.to_lower_hex_string()
),
Err(existing_state_id) => log::debug!("State {} already exists", existing_state_id),
}
if commit_msg.validation_tokens.len() > 0 {
log::debug!(
"Received commit_msg with {} validation tokens for process {}",
commit_msg.validation_tokens.len(),
process_id
);
// If the validation succeed, we return a new tip
process_validation(process_to_udpate, commit_msg)?;
}
Ok(())
}
pub fn dump_cached_members() -> Result<(), anyhow::Error> {
let members = lock_members()?.clone();
let storage = STORAGE
.get()
.ok_or(Error::msg("STORAGE is not initialized"))?
.lock_anyhow()?;
let members_file = &storage.members_file;
let members_map = OutPointMemberMap(members);
let json = serde_json::to_value(&members_map)?;
members_file.save(&json)?;
log::debug!("saved members");
Ok(())
}
pub fn dump_cached_processes(processes: HashMap<OutPoint, Process>) -> Result<(), anyhow::Error> {
let storage = STORAGE
.get()
.ok_or(Error::msg("STORAGE is not initialized"))?
.lock_anyhow()?;
let processes_file = &storage.processes_file;
let outpoints_map = OutPointProcessMap(processes);
let json = serde_json::to_value(&outpoints_map)?;
processes_file.save(&json)?;
log::debug!("saved processes");
Ok(())
}
// Register a new state
fn register_new_state(process: &mut Process, commit_msg: &CommitMessage) -> Result<[u8; 32]> {
let last_commited_state = process.get_latest_commited_state();
let new_state_id = commit_msg
.pcd_commitment
.create_merkle_tree()?
.root()
.unwrap();
if let Some(state) = last_commited_state {
if new_state_id == state.state_id {
return Err(Error::msg(format!(
"{}",
new_state_id.to_lower_hex_string()
)));
}
}
let concurrent_states = process.get_latest_concurrent_states()?;
let (empty_state, actual_states) = concurrent_states.split_last().unwrap();
let current_outpoint = empty_state.commited_in;
// Ensure no duplicate states
if actual_states
.iter()
.any(|state| state.state_id == new_state_id)
{
return Err(Error::msg(format!(
"{}",
new_state_id.to_lower_hex_string()
)));
}
// Add the new state
let new_state = ProcessState {
commited_in: current_outpoint,
pcd_commitment: commit_msg.pcd_commitment.clone(),
state_id: new_state_id.clone(),
roles: commit_msg.roles.clone(),
public_data: commit_msg.public_data.clone(),
..Default::default()
};
process.insert_concurrent_state(new_state)?;
Ok(new_state_id)
}
// Process validation for a state with validation tokens
fn process_validation(
updated_process: &mut Process,
commit_msg: &CommitMessage,
) -> Result<OutPoint> {
let new_state_id = if commit_msg.pcd_commitment.is_empty() {
// We're dealing with an obliteration attempt
[0u8; 32]
} else {
commit_msg
.pcd_commitment
.create_merkle_tree()?
.root()
.ok_or(Error::msg("Invalid merkle tree"))?
};
{
let state_to_update = updated_process.get_state_for_id_mut(&new_state_id)?;
// Complete with the received tokens
state_to_update
.validation_tokens
.extend(commit_msg.validation_tokens.iter());
state_to_update.validation_tokens.sort_unstable();
state_to_update.validation_tokens.dedup();
}
let state_to_validate = updated_process.get_state_for_id(&new_state_id)?;
let members = lock_members()?.clone();
state_to_validate.is_valid(
updated_process.get_latest_commited_state(),
&OutPointMemberMap(members),
)?;
let commited_in = commit_new_transaction(updated_process, state_to_validate.clone())?;
Ok(commited_in)
}
// Commit the new transaction and update the process state
fn commit_new_transaction(
updated_process: &mut Process,
state_to_commit: ProcessState,
) -> Result<OutPoint> {
let sp_wallet = WALLET
.get()
.ok_or(Error::msg("Wallet not initialized"))?
.lock_anyhow()?;
let commitment_payload = Vec::from(state_to_commit.state_id);
let mut recipients = vec![];
recipients.push(Recipient {
address: RecipientAddress::SpAddress(sp_wallet.get_sp_client().get_receiving_address()),
amount: Amount::from_sat(1000),
});
// TODO not sure if this is still used
// If the process is a pairing, we add another output that directly pays the owner of the process
// We can find out simply by looking at the members list
if let Some(member) = lock_members()?.get(&updated_process.get_process_id().unwrap()) {
// We just pick one of the devices of the member at random en pay to it, member can then share the private key between all devices
// For now we take the first address
let address: SilentPaymentAddress =
member.get_addresses().get(0).unwrap().as_str().try_into()?;
recipients.push(Recipient {
address: RecipientAddress::SpAddress(address),
amount: Amount::from_sat(1000),
});
}
// This output is used to generate publicly available public keys without having to go through too many loops
let daemon = DAEMON.get().unwrap().lock_anyhow()?;
let fee_rate = daemon
.estimate_fee(6)
.unwrap_or(Amount::from_sat(1000))
.checked_div(1000)
.unwrap();
let mut freezed_utxos = lock_freezed_utxos()?;
let next_commited_in = updated_process.get_process_tip()?;
if !freezed_utxos.contains(&next_commited_in) {
return Err(Error::msg(format!(
"Missing next commitment outpoint for process {}",
updated_process.get_process_id()?
)));
};
let unspent_outputs = sp_wallet.get_unspent_outputs();
let mut available_outpoints = vec![];
// We push the next_commited_in at the top of the available outpoints
if let Some(output) = unspent_outputs.get(&next_commited_in) {
available_outpoints.push((next_commited_in, output.clone()));
}
// We filter out freezed utxos
for (outpoint, output) in unspent_outputs {
if !freezed_utxos.contains(&outpoint) {
available_outpoints.push((outpoint, output));
}
}
let unsigned_transaction = create_transaction(
available_outpoints,
sp_wallet.get_sp_client(),
recipients,
Some(commitment_payload),
FeeRate::from_sat_per_vb(fee_rate.to_sat() as f32),
)?;
let final_tx = sign_transaction(sp_wallet.get_sp_client(), unsigned_transaction)?;
daemon.test_mempool_accept(&final_tx)?;
let txid = daemon.broadcast(&final_tx)?;
let commited_in = OutPoint::new(txid, 0);
freezed_utxos.insert(commited_in);
freezed_utxos.remove(&next_commited_in);
updated_process.remove_all_concurrent_states()?;
updated_process.insert_concurrent_state(state_to_commit)?;
updated_process.update_states_tip(commited_in)?;
Ok(commited_in)
}
// TODO tests are broken, we need a complete overhaul to make it work again
#[cfg(test)]
mod tests {
use super::*;
use crate::daemon::RpcCall;
use crate::DiskStorage;
use crate::StateFile;
use bitcoincore_rpc::bitcoin::consensus::deserialize;
use bitcoincore_rpc::bitcoin::hex::DisplayHex;
use bitcoincore_rpc::bitcoin::*;
use mockall::mock;
use mockall::predicate::*;
use sdk_common::pcd::Member;
use sdk_common::pcd::Pcd;
use sdk_common::pcd::PcdCommitments;
use sdk_common::pcd::RoleDefinition;
use sdk_common::pcd::Roles;
use sdk_common::pcd::ValidationRule;
use sdk_common::process::CACHEDPROCESSES;
use sdk_common::sp_client::bitcoin::consensus::serialize;
use sdk_common::sp_client::bitcoin::hex::FromHex;
use sdk_common::sp_client::silentpayments::SilentPaymentAddress;
use serde_json::json;
use serde_json::{Map, Value};
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Mutex;
use std::sync::OnceLock;
const LOCAL_ADDRESS: &str = "sprt1qq222dhaxlzmjft2pa7qtspw2aw55vwfmtnjyllv5qrsqwm3nufxs6q7t88jf9asvd7rxhczt87de68du3jhem54xvqxy80wc6ep7lauxacsrq79v";
const INIT_TRANSACTION: &str = "02000000000102b01b832bf34cf87583c628839c5316546646dcd4939e339c1d83e693216cdfa00100000000fdffffffdd1ca865b199accd4801634488fca87e0cf81b36ee7e9bec526a8f922539b8670000000000fdffffff0200e1f505000000001600140798fac9f310cefad436ea928f0bdacf03a11be544e0f5050000000016001468a66f38e7c2c9e367577d6fad8532ae2c728ed2014043764b77de5041f80d19e3d872f205635f87486af015c00d2a3b205c694a0ae1cbc60e70b18bcd4470abbd777de63ae52600aba8f5ad1334cdaa6bcd931ab78b0140b56dd8e7ac310d6dcbc3eff37f111ced470990d911b55cd6ff84b74b579c17d0bba051ec23b738eeeedba405a626d95f6bdccb94c626db74c57792254bfc5a7c00000000";
const TMP_WALLET: &str = "/tmp/.4nk/wallet";
const TMP_PROCESSES: &str = "/tmp/.4nk/processes";
const TMP_MEMBERS: &str = "/tmp/.4nk/members";
// Define the mock for Daemon with the required methods
mock! {
#[derive(Debug)]
pub Daemon {}
impl RpcCall for Daemon {
fn connect(
rpcwallet: Option<String>,
rpc_url: String,
network: bitcoincore_rpc::bitcoin::Network,
) -> Result<Self> where Self: Sized;
fn estimate_fee(&self, nblocks: u16) -> Result<Amount>;
fn get_relay_fee(&self) -> Result<Amount>;
fn get_current_height(&self) -> Result<u64>;
fn get_block(&self, block_hash: BlockHash) -> Result<Block>;
fn get_filters(&self, block_height: u32) -> Result<(u32, BlockHash, bip158::BlockFilter)>;
fn list_unspent_from_to(
&self,
minamt: Option<Amount>,
) -> Result<Vec<bitcoincore_rpc::json::ListUnspentResultEntry>>;
fn create_psbt(
&self,
unspents: &[bitcoincore_rpc::json::ListUnspentResultEntry],
spk: ScriptBuf,
network: Network,
) -> Result<String>;
fn process_psbt(&self, psbt: String) -> Result<String>;
fn finalize_psbt(&self, psbt: String) -> Result<String>;
fn get_network(&self) -> Result<Network>;
fn test_mempool_accept(
&self,
tx: &Transaction,
) -> Result<crate::bitcoin_json::TestMempoolAcceptResult>;
fn broadcast(&self, tx: &Transaction) -> Result<Txid>;
fn get_transaction_info(
&self,
txid: &Txid,
blockhash: Option<BlockHash>,
) -> Result<Value>;
fn get_transaction_hex(
&self,
txid: &Txid,
blockhash: Option<BlockHash>,
) -> Result<Value>;
fn get_transaction(
&self,
txid: &Txid,
blockhash: Option<BlockHash>,
) -> Result<Transaction>;
fn get_block_txids(&self, blockhash: BlockHash) -> Result<Vec<Txid>>;
fn get_mempool_txids(&self) -> Result<Vec<Txid>>;
fn get_mempool_entries(
&self,
txids: &[Txid],
) -> Result<Vec<Result<bitcoincore_rpc::json::GetMempoolEntryResult>>>;
fn get_mempool_transactions(
&self,
txids: &[Txid],
) -> Result<Vec<Result<Transaction>>>;
}
}
mock! {
#[derive(Debug)]
pub SpWallet {
fn get_receiving_address(&self) -> Result<String>;
}
}
mock! {
#[derive(Debug)]
pub SilentPaymentWallet {
fn get_sp_wallet(&self) -> Result<MockSpWallet>;
}
}
static WALLET: OnceLock<MockSilentPaymentWallet> = OnceLock::new();
pub fn initialize_static_variables() {
if DAEMON.get().is_none() {
let mut daemon = MockDaemon::new();
daemon
.expect_broadcast()
.withf(|tx: &Transaction| serialize(tx).to_lower_hex_string() == INIT_TRANSACTION)
.returning(|tx| Ok(tx.txid()));
DAEMON
.set(Mutex::new(Box::new(daemon)))
.expect("DAEMON should only be initialized once");
println!("Initialized DAEMON");
}
if WALLET.get().is_none() {
let mut wallet = MockSilentPaymentWallet::new();
wallet
.expect_get_sp_wallet()
.returning(|| Ok(MockSpWallet::new()));
WALLET
.set(wallet)
.expect("WALLET should only be initialized once");
println!("Initialized WALLET");
}
if CACHEDPROCESSES.get().is_none() {
CACHEDPROCESSES
.set(Mutex::new(HashMap::new()))
.expect("CACHEDPROCESSES should only be initialized once");
println!("Initialized CACHEDPROCESSES");
}
if STORAGE.get().is_none() {
let wallet_file = StateFile::new(PathBuf::from_str(TMP_WALLET).unwrap());
let processes_file = StateFile::new(PathBuf::from_str(TMP_PROCESSES).unwrap());
let members_file = StateFile::new(PathBuf::from_str(TMP_MEMBERS).unwrap());
wallet_file.create().unwrap();
processes_file.create().unwrap();
members_file.create().unwrap();
let disk_storage = DiskStorage {
wallet_file,
processes_file,
members_file,
};
STORAGE
.set(Mutex::new(disk_storage))
.expect("STORAGE should initialize only once");
println!("Initialized STORAGE");
}
}
fn mock_commit_msg(process_id: OutPoint) -> CommitMessage {
let field_names = [
"a".to_owned(),
"b".to_owned(),
"pub_a".to_owned(),
"roles".to_owned(),
];
let pairing_id = OutPoint::from_str(
"b0c8378ee68e9a73836b04423ddb6de9fc0e2e715e04ffe6aa34117bb1025f01:0",
)
.unwrap();
let member = Member::new(vec![SilentPaymentAddress::try_from(LOCAL_ADDRESS).unwrap()]);
let validation_rule = ValidationRule::new(1.0, Vec::from(field_names), 1.0).unwrap();
let role_def = RoleDefinition {
members: vec![pairing_id],
validation_rules: vec![validation_rule],
storages: vec![],
};
let roles = Roles::new(BTreeMap::from([(String::from("role_name"), role_def)]));
let public_data = TryInto::<Pcd>::try_into(json!({"pub_a": Value::Null})).unwrap();
let clear_state =
TryInto::<Pcd>::try_into(json!({"a": Value::Null, "b": Value::Null})).unwrap();
let pcd_commitments = PcdCommitments::new(
&process_id,
&Pcd::new(public_data.clone().into_iter().chain(clear_state).collect()),
&roles,
)
.unwrap();
let commit_msg = CommitMessage {
process_id,
roles,
public_data,
validation_tokens: vec![],
pcd_commitment: pcd_commitments,
error: None,
};
commit_msg
}
#[test]
fn test_handle_commit_new_process() {
initialize_static_variables();
let init_tx =
deserialize::<Transaction>(&Vec::from_hex(INIT_TRANSACTION).unwrap()).unwrap();
let init_txid = init_tx.txid();
let process_id = OutPoint::new(init_txid, 0);
let commit_msg = mock_commit_msg(process_id);
let roles = commit_msg.roles.clone();
let pcd_commitment = commit_msg.pcd_commitment.clone();
let empty_state = ProcessState {
commited_in: process_id,
..Default::default()
};
let result = handle_commit_request(commit_msg);
assert_eq!(result.unwrap(), process_id);
let cache = CACHEDPROCESSES.get().unwrap().lock().unwrap();
let updated_process = cache.get(&process_id);
assert!(updated_process.is_some());
let concurrent_states = updated_process
.unwrap()
.get_latest_concurrent_states()
.unwrap();
// Constructing the roles_map that was inserted in the process
let roles_object = serde_json::to_value(roles).unwrap();
let mut roles_map = Map::new();
roles_map.insert("roles".to_owned(), roles_object);
let new_state = ProcessState {
commited_in: process_id,
pcd_commitment,
..Default::default()
};
let target = vec![&empty_state, &new_state];
assert_eq!(concurrent_states, target);
}
#[test]
fn test_handle_commit_new_state() {
initialize_static_variables();
let init_tx =
deserialize::<Transaction>(&Vec::from_hex(INIT_TRANSACTION).unwrap()).unwrap();
let init_txid = init_tx.txid();
let process_id = OutPoint::new(init_txid, 0);
let commit_msg = mock_commit_msg(process_id);
let roles = commit_msg.roles.clone();
let pcd_commitment = commit_msg.pcd_commitment.clone();
let process = Process::new(process_id);
CACHEDPROCESSES
.get()
.unwrap()
.lock()
.unwrap()
.insert(process_id, process);
let result = handle_commit_request(commit_msg);
assert_eq!(result.unwrap(), process_id);
let cache = CACHEDPROCESSES.get().unwrap().lock().unwrap();
let updated_process = cache.get(&process_id);
assert!(updated_process.is_some());
let concurrent_states = updated_process
.unwrap()
.get_latest_concurrent_states()
.unwrap();
let roles_object = serde_json::to_value(roles).unwrap();
let mut roles_map = Map::new();
roles_map.insert("roles".to_owned(), roles_object);
let new_state = ProcessState {
commited_in: process_id,
pcd_commitment,
..Default::default()
};
let empty_state = ProcessState {
commited_in: process_id,
..Default::default()
};
let target = vec![&empty_state, &new_state];
assert_eq!(concurrent_states, target);
}
// #[test]
// fn test_handle_commit_request_invalid_init_tx() {
// let commit_msg = CommitMessage {
// init_tx: "invalid_tx_hex".to_string(),
// roles: HashMap::new(),
// validation_tokens: vec![],
// pcd_commitment: json!({"roles": "expected_roles"}).as_object().unwrap().clone(),
// };
// // Call the function under test
// let result = handle_commit_request(commit_msg);
// // Assertions for error
// assert!(result.is_err());
// assert_eq!(result.unwrap_err().to_string(), "init_tx must be a valid transaction or txid");
// }
// // Example test for adding a new state to an existing commitment
// #[test]
// fn test_handle_commit_request_add_state() {
// // Set up data for adding a state to an existing commitment
// let commit_msg = CommitMessage {
// init_tx: "existing_outpoint_hex".to_string(),
// roles: HashMap::new(),
// validation_tokens: vec![],
// pcd_commitment: json!({"roles": "expected_roles"}).as_object().unwrap().clone(),
// };
// // Mock daemon and cache initialization
// let mut daemon = MockDaemon::new();
// daemon.expect_broadcast().returning(|_| Ok(Txid::new()));
// DAEMON.set(Arc::new(Mutex::new(daemon))).unwrap();
// let process_state = Process::new(vec![], vec![]);
// CACHEDPROCESSES.lock().unwrap().insert(OutPoint::new("mock_txid", 0), process_state);
// // Run the function
// let result = handle_commit_request(commit_msg);
// // Assert success and that a new state was added
// assert!(result.is_ok());
// assert_eq!(result.unwrap(), OutPoint::new("mock_txid", 0));
// }
// // Additional tests for errors and validation tokens would follow a similar setup
}

96
src/config.rs
View File

@ -1,96 +0,0 @@
use std::collections::HashMap;
use std::fs::File;
use std::io::{self, BufRead};
use anyhow::{Error, Result};
use sdk_common::sp_client::bitcoin::Network;
#[derive(Debug)]
pub struct Config {
pub core_url: String,
pub core_wallet: Option<String>,
pub ws_url: String,
pub http_url: String,
pub wallet_name: String,
pub network: Network,
pub electrum_url: String,
pub zmq_url: String,
pub data_dir: String,
pub idnot_client_secret: Option<String>,
pub idnot_client_id: Option<String>,
pub idnot_redirect_uri: Option<String>,
}
impl Config {
pub fn read_from_file(filename: &str) -> Result<Self> {
let mut file_content = HashMap::new();
if let Ok(file) = File::open(filename) {
let reader = io::BufReader::new(file);
// Read the file line by line
for line in reader.lines() {
if let Ok(l) = line {
// Ignore comments and empty lines
if l.starts_with('#') || l.trim().is_empty() {
continue;
}
// Split the line into key and value
if let Some((k, v)) = l.split_once('=') {
file_content.insert(k.to_owned(), v.trim_matches('\"').to_owned());
}
}
}
} else {
return Err(anyhow::Error::msg("Failed to find conf file"));
}
// Now set the Config
let config = Config {
core_url: file_content
.remove("core_url")
.ok_or(Error::msg("No \"core_url\""))?
.to_owned(),
core_wallet: file_content.remove("core_wallet").map(|s| s.to_owned()),
ws_url: file_content
.remove("ws_url")
.ok_or(Error::msg("No \"ws_url\""))?
.to_owned(),
http_url: file_content
.remove("http_url")
.ok_or(Error::msg("No \"http_url\""))?
.to_owned(),
wallet_name: file_content
.remove("wallet_name")
.ok_or(Error::msg("No \"wallet_name\""))?
.to_owned(),
network: Network::from_core_arg(
&file_content
.remove("network")
.ok_or(Error::msg("no \"network\""))?
.trim_matches('\"'),
)?,
electrum_url: file_content
.remove("electrum_url")
.ok_or(Error::msg("No \"electrum_url\""))?
.to_owned(),
zmq_url: file_content
.remove("zmq_url")
.ok_or(Error::msg("No \"zmq_url\""))?
.to_owned(),
data_dir: file_content
.remove("data_dir")
.ok_or(Error::msg("No \"data_dir\""))?
.to_owned(),
idnot_client_id: file_content
.remove("idnot_client_id"),
idnot_client_secret: file_content
.remove("idnot_client_secret"),
idnot_redirect_uri: file_content
.remove("idnot_redirect_uri"),
};
Ok(config)
}
}

460
src/daemon.rs
View File

@ -1,460 +0,0 @@
use anyhow::{Context, Error, Result};
use bitcoincore_rpc::json::{
CreateRawTransactionInput, ListUnspentQueryOptions, ListUnspentResultEntry,
WalletCreateFundedPsbtOptions,
};
use bitcoincore_rpc::{json, jsonrpc, Auth, Client, RpcApi};
use sdk_common::sp_client::bitcoin::bip158::BlockFilter;
use sdk_common::sp_client::bitcoin::{
block, Address, Amount, Block, BlockHash, Network, OutPoint, Psbt, ScriptBuf, Sequence,
Transaction, TxIn, TxOut, Txid,
};
use sdk_common::sp_client::bitcoin::{consensus::deserialize, hashes::hex::FromHex};
// use crossbeam_channel::Receiver;
// use parking_lot::Mutex;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::env;
use std::fs::File;
use std::io::Read;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::time::Duration;
use crate::FAUCET_AMT;
pub struct SensitiveAuth(pub Auth);
impl SensitiveAuth {
pub(crate) fn get_auth(&self) -> Auth {
self.0.clone()
}
}
enum PollResult {
Done(Result<()>),
Retry,
}
fn rpc_poll(client: &mut Client, skip_block_download_wait: bool) -> PollResult {
match client.get_blockchain_info() {
Ok(info) => {
if skip_block_download_wait {
// bitcoind RPC is available, don't wait for block download to finish
return PollResult::Done(Ok(()));
}
let left_blocks = info.headers - info.blocks;
if info.initial_block_download || left_blocks > 0 {
log::info!(
"waiting for {} blocks to download{}",
left_blocks,
if info.initial_block_download {
" (IBD)"
} else {
""
}
);
return PollResult::Retry;
}
PollResult::Done(Ok(()))
}
Err(err) => {
if let Some(e) = extract_bitcoind_error(&err) {
if e.code == -28 {
log::debug!("waiting for RPC warmup: {}", e.message);
return PollResult::Retry;
}
}
PollResult::Done(Err(err).context("daemon not available"))
}
}
}
fn read_cookie(path: &Path) -> Result<(String, String)> {
// Load username and password from bitcoind cookie file:
// * https://github.com/bitcoin/bitcoin/pull/6388/commits/71cbeaad9a929ba6a7b62d9b37a09b214ae00c1a
// * https://bitcoin.stackexchange.com/questions/46782/rpc-cookie-authentication
let mut file = File::open(path)
.with_context(|| format!("failed to open bitcoind cookie file: {}", path.display()))?;
let mut contents = String::new();
file.read_to_string(&mut contents)
.with_context(|| format!("failed to read bitcoind cookie from {}", path.display()))?;
let parts: Vec<&str> = contents.splitn(2, ':').collect();
anyhow::ensure!(
parts.len() == 2,
"failed to parse bitcoind cookie - missing ':' separator"
);
Ok((parts[0].to_owned(), parts[1].to_owned()))
}
fn rpc_connect(rpcwallet: Option<String>, network: Network, mut rpc_url: String) -> Result<Client> {
match rpcwallet {
Some(rpcwallet) => rpc_url.push_str(&rpcwallet),
None => (),
}
// Allow `wait_for_new_block` to take a bit longer before timing out.
// See https://github.com/romanz/electrs/issues/495 for more details.
let builder = jsonrpc::simple_http::SimpleHttpTransport::builder()
.url(&rpc_url)?
.timeout(Duration::from_secs(30));
let home = env::var("HOME")?;
let mut cookie_path = PathBuf::from_str(&home)?;
cookie_path.push(".bitcoin");
cookie_path.push(network.to_core_arg());
cookie_path.push(".cookie");
let daemon_auth = SensitiveAuth(Auth::CookieFile(cookie_path));
let builder = match daemon_auth.get_auth() {
Auth::None => builder,
Auth::UserPass(user, pass) => builder.auth(user, Some(pass)),
Auth::CookieFile(path) => {
let (user, pass) = read_cookie(&path)?;
builder.auth(user, Some(pass))
}
};
Ok(Client::from_jsonrpc(jsonrpc::Client::with_transport(
builder.build(),
)))
}
#[derive(Debug)]
pub struct Daemon {
rpc: Client,
}
impl RpcCall for Daemon {
fn connect(rpcwallet: Option<String>, rpc_url: String, network: Network) -> Result<Self> {
let mut rpc = rpc_connect(rpcwallet, network, rpc_url)?;
loop {
match rpc_poll(&mut rpc, false) {
PollResult::Done(result) => {
result.context("bitcoind RPC polling failed")?;
break; // on success, finish polling
}
PollResult::Retry => {
std::thread::sleep(std::time::Duration::from_secs(1)); // wait a bit before polling
}
}
}
let network_info = rpc.get_network_info()?;
if !network_info.network_active {
anyhow::bail!("electrs requires active bitcoind p2p network");
}
let info = rpc.get_blockchain_info()?;
if info.pruned {
anyhow::bail!("electrs requires non-pruned bitcoind node");
}
Ok(Self { rpc })
}
fn estimate_fee(&self, nblocks: u16) -> Result<Amount> {
let res = self
.rpc
.estimate_smart_fee(nblocks, None)
.context("failed to estimate fee")?;
if res.errors.is_some() {
Err(Error::msg(serde_json::to_string(&res.errors.unwrap())?))
} else {
Ok(res.fee_rate.unwrap())
}
}
fn get_relay_fee(&self) -> Result<Amount> {
Ok(self
.rpc
.get_network_info()
.context("failed to get relay fee")?
.relay_fee)
}
fn get_current_height(&self) -> Result<u64> {
Ok(self
.rpc
.get_block_count()
.context("failed to get block count")?)
}
fn get_block(&self, block_hash: BlockHash) -> Result<Block> {
Ok(self
.rpc
.get_block(&block_hash)
.context("failed to get block")?)
}
fn get_filters(&self, block_height: u32) -> Result<(u32, BlockHash, BlockFilter)> {
let block_hash = self.rpc.get_block_hash(block_height.try_into()?)?;
let filter = self
.rpc
.get_block_filter(&block_hash)
.context("failed to get block filter")?
.into_filter();
Ok((block_height, block_hash, filter))
}
fn list_unspent_from_to(
&self,
minamt: Option<Amount>,
) -> Result<Vec<json::ListUnspentResultEntry>> {
let minimum_sum_amount = if minamt.is_none() || minamt <= FAUCET_AMT.checked_mul(2) {
FAUCET_AMT.checked_mul(2)
} else {
minamt
};
Ok(self.rpc.list_unspent(
None,
None,
None,
Some(true),
Some(ListUnspentQueryOptions {
minimum_sum_amount,
..Default::default()
}),
)?)
}
fn create_psbt(
&self,
unspents: &[ListUnspentResultEntry],
spk: ScriptBuf,
network: Network,
) -> Result<String> {
let inputs: Vec<CreateRawTransactionInput> = unspents
.iter()
.map(|utxo| CreateRawTransactionInput {
txid: utxo.txid,
vout: utxo.vout,
sequence: None,
})
.collect();
let address = Address::from_script(&spk, network)?;
let total_amt = unspents
.iter()
.fold(Amount::from_sat(0), |acc, x| acc + x.amount);
if total_amt < FAUCET_AMT {
return Err(Error::msg("Not enought funds"));
}
let mut outputs = HashMap::new();
outputs.insert(address.to_string(), total_amt);
let options = WalletCreateFundedPsbtOptions {
subtract_fee_from_outputs: vec![0],
..Default::default()
};
let wallet_create_funded_result =
self.rpc
.wallet_create_funded_psbt(&inputs, &outputs, None, Some(options), None)?;
Ok(wallet_create_funded_result.psbt.to_string())
}
fn process_psbt(&self, psbt: String) -> Result<String> {
let processed_psbt = self.rpc.wallet_process_psbt(&psbt, None, None, None)?;
match processed_psbt.complete {
true => Ok(processed_psbt.psbt),
false => Err(Error::msg("Failed to complete the psbt")),
}
}
fn finalize_psbt(&self, psbt: String) -> Result<String> {
let final_tx = self.rpc.finalize_psbt(&psbt, Some(false))?;
match final_tx.complete {
true => Ok(final_tx
.psbt
.expect("We shouldn't have an empty psbt for a complete return")),
false => Err(Error::msg("Failed to finalize psbt")),
}
}
fn get_network(&self) -> Result<Network> {
let blockchain_info = self.rpc.get_blockchain_info()?;
Ok(blockchain_info.chain)
}
fn test_mempool_accept(
&self,
tx: &Transaction,
) -> Result<crate::bitcoin_json::TestMempoolAcceptResult> {
let res = self.rpc.test_mempool_accept(&vec![tx])?;
Ok(res.get(0).unwrap().clone())
}
fn broadcast(&self, tx: &Transaction) -> Result<Txid> {
let txid = self.rpc.send_raw_transaction(tx)?;
Ok(txid)
}
fn get_transaction_info(&self, txid: &Txid, blockhash: Option<BlockHash>) -> Result<Value> {
// No need to parse the resulting JSON, just return it as-is to the client.
self.rpc
.call(
"getrawtransaction",
&[json!(txid), json!(true), json!(blockhash)],
)
.context("failed to get transaction info")
}
fn get_transaction_hex(&self, txid: &Txid, blockhash: Option<BlockHash>) -> Result<Value> {
use sdk_common::sp_client::bitcoin::consensus::serde::{hex::Lower, Hex, With};
let tx = self.get_transaction(txid, blockhash)?;
#[derive(serde::Serialize)]
#[serde(transparent)]
struct TxAsHex(#[serde(with = "With::<Hex<Lower>>")] Transaction);
serde_json::to_value(TxAsHex(tx)).map_err(Into::into)
}
fn get_transaction(&self, txid: &Txid, blockhash: Option<BlockHash>) -> Result<Transaction> {
self.rpc
.get_raw_transaction(txid, blockhash.as_ref())
.context("failed to get transaction")
}
fn get_block_txids(&self, blockhash: BlockHash) -> Result<Vec<Txid>> {
Ok(self
.rpc
.get_block_info(&blockhash)
.context("failed to get block txids")?
.tx)
}
fn get_mempool_txids(&self) -> Result<Vec<Txid>> {
self.rpc
.get_raw_mempool()
.context("failed to get mempool txids")
}
fn get_mempool_entries(
&self,
txids: &[Txid],
) -> Result<Vec<Result<json::GetMempoolEntryResult>>> {
let client = self.rpc.get_jsonrpc_client();
log::debug!("getting {} mempool entries", txids.len());
let args: Vec<_> = txids
.iter()
.map(|txid| vec![serde_json::value::to_raw_value(txid).unwrap()])
.collect();
let reqs: Vec<_> = args
.iter()
.map(|a| client.build_request("getmempoolentry", a))
.collect();
let res = client.send_batch(&reqs).context("batch request failed")?;
log::debug!("got {} mempool entries", res.len());
Ok(res
.into_iter()
.map(|r| {
r.context("missing response")?
.result::<json::GetMempoolEntryResult>()
.context("invalid response")
})
.collect())
}
fn get_mempool_transactions(&self, txids: &[Txid]) -> Result<Vec<Result<Transaction>>> {
let client = self.rpc.get_jsonrpc_client();
log::debug!("getting {} transactions", txids.len());
let args: Vec<_> = txids
.iter()
.map(|txid| vec![serde_json::value::to_raw_value(txid).unwrap()])
.collect();
let reqs: Vec<_> = args
.iter()
.map(|a| client.build_request("getrawtransaction", a))
.collect();
let res = client.send_batch(&reqs).context("batch request failed")?;
log::debug!("got {} mempool transactions", res.len());
Ok(res
.into_iter()
.map(|r| -> Result<Transaction> {
let tx_hex = r
.context("missing response")?
.result::<String>()
.context("invalid response")?;
let tx_bytes = Vec::from_hex(&tx_hex).context("non-hex transaction")?;
deserialize(&tx_bytes).context("invalid transaction")
})
.collect())
}
}
pub(crate) trait RpcCall: Send + Sync + std::fmt::Debug {
fn connect(rpcwallet: Option<String>, rpc_url: String, network: Network) -> Result<Self>
where
Self: Sized;
fn estimate_fee(&self, nblocks: u16) -> Result<Amount>;
fn get_relay_fee(&self) -> Result<Amount>;
fn get_current_height(&self) -> Result<u64>;
fn get_block(&self, block_hash: BlockHash) -> Result<Block>;
fn get_filters(&self, block_height: u32) -> Result<(u32, BlockHash, BlockFilter)>;
fn list_unspent_from_to(
&self,
minamt: Option<Amount>,
) -> Result<Vec<json::ListUnspentResultEntry>>;
fn create_psbt(
&self,
unspents: &[ListUnspentResultEntry],
spk: ScriptBuf,
network: Network,
) -> Result<String>;
fn process_psbt(&self, psbt: String) -> Result<String>;
fn finalize_psbt(&self, psbt: String) -> Result<String>;
fn get_network(&self) -> Result<Network>;
fn test_mempool_accept(
&self,
tx: &Transaction,
) -> Result<crate::bitcoin_json::TestMempoolAcceptResult>;
fn broadcast(&self, tx: &Transaction) -> Result<Txid>;
fn get_transaction_info(&self, txid: &Txid, blockhash: Option<BlockHash>) -> Result<Value>;
fn get_transaction_hex(&self, txid: &Txid, blockhash: Option<BlockHash>) -> Result<Value>;
fn get_transaction(&self, txid: &Txid, blockhash: Option<BlockHash>) -> Result<Transaction>;
fn get_block_txids(&self, blockhash: BlockHash) -> Result<Vec<Txid>>;
fn get_mempool_txids(&self) -> Result<Vec<Txid>>;
fn get_mempool_entries(
&self,
txids: &[Txid],
) -> Result<Vec<Result<json::GetMempoolEntryResult>>>;
fn get_mempool_transactions(&self, txids: &[Txid]) -> Result<Vec<Result<Transaction>>>;
}
pub(crate) type RpcError = bitcoincore_rpc::jsonrpc::error::RpcError;
pub(crate) fn extract_bitcoind_error(err: &bitcoincore_rpc::Error) -> Option<&RpcError> {
use bitcoincore_rpc::{
jsonrpc::error::Error::Rpc as ServerError, Error::JsonRpc as JsonRpcError,
};
match err {
JsonRpcError(ServerError(e)) => Some(e),
_ => None,
}
}

14
src/electrumclient.rs
View File

@ -1,14 +0,0 @@
use electrum_client::{Client, ConfigBuilder};
use log::info;
const VALIDATE_DOMAIN: bool = false; // self-signed cert, so we don't validate
pub fn create_electrum_client(electrum_url: &str) -> anyhow::Result<Client> {
let config = ConfigBuilder::new()
.validate_domain(VALIDATE_DOMAIN)
.build();
let electrum_client = Client::from_config(electrum_url, config)?;
info!("ssl client {}", electrum_url);
Ok(electrum_client)
}

274
src/faucet.rs
View File

@ -1,274 +0,0 @@
use std::{collections::HashMap, str::FromStr};
use bitcoincore_rpc::bitcoin::secp256k1::PublicKey;
use bitcoincore_rpc::json::{self as bitcoin_json};
use sdk_common::silentpayments::sign_transaction;
use sdk_common::sp_client::bitcoin::secp256k1::{
rand::thread_rng, Keypair, Message as Secp256k1Message, Secp256k1, ThirtyTwoByteHash,
};
use sdk_common::sp_client::bitcoin::{
absolute::LockTime,
consensus::serialize,
hex::{DisplayHex, FromHex},
key::TapTweak,
script::PushBytesBuf,
sighash::{Prevouts, SighashCache},
taproot::Signature,
transaction::Version,
Amount, OutPoint, Psbt, ScriptBuf, TapSighashType, Transaction, TxIn, TxOut, Witness,
XOnlyPublicKey,
};
use sdk_common::{
network::{FaucetMessage, NewTxMessage},
silentpayments::create_transaction,
};
use sdk_common::sp_client::silentpayments::sending::generate_recipient_pubkeys;
use sdk_common::sp_client::silentpayments::utils::sending::calculate_partial_secret;
use sdk_common::sp_client::{FeeRate, OwnedOutput, Recipient, RecipientAddress};
use anyhow::{Error, Result};
use crate::lock_freezed_utxos;
use crate::scan::check_transaction_alone;
use crate::{
scan::compute_partial_tweak_to_transaction, MutexExt, SilentPaymentAddress, DAEMON, FAUCET_AMT,
WALLET,
};
fn spend_from_core(dest: XOnlyPublicKey) -> Result<(Transaction, Amount)> {
let core = DAEMON
.get()
.ok_or(Error::msg("DAEMON not initialized"))?
.lock_anyhow()?;
let unspent_list: Vec<bitcoin_json::ListUnspentResultEntry> =
core.list_unspent_from_to(None)?;
if !unspent_list.is_empty() {
let network = core.get_network()?;
let spk = ScriptBuf::new_p2tr_tweaked(dest.dangerous_assume_tweaked());
let new_psbt = core.create_psbt(&unspent_list, spk, network)?;
let processed_psbt = core.process_psbt(new_psbt)?;
let finalize_psbt_result = core.finalize_psbt(processed_psbt)?;
let final_psbt = Psbt::from_str(&finalize_psbt_result)?;
let total_fee = final_psbt.fee()?;
let final_tx = final_psbt.extract_tx()?;
let fee_rate = total_fee
.checked_div(final_tx.weight().to_vbytes_ceil())
.unwrap();
Ok((final_tx, fee_rate))
} else {
// we don't have enough available coins to pay for this faucet request
Err(Error::msg("No spendable outputs"))
}
}
fn faucet_send(
sp_address: SilentPaymentAddress,
commitment: &str,
) -> Result<(Transaction, PublicKey)> {
let sp_wallet = WALLET
.get()
.ok_or(Error::msg("Wallet not initialized"))?
.lock_anyhow()?;
let fee_estimate = DAEMON
.get()
.ok_or(Error::msg("DAEMON not initialized"))?
.lock_anyhow()?
.estimate_fee(6)
.unwrap_or(Amount::from_sat(1000))
.checked_div(1000)
.unwrap();
log::debug!("fee estimate for 6 blocks: {}", fee_estimate);
let recipient = Recipient {
address: RecipientAddress::SpAddress(sp_address),
amount: FAUCET_AMT,
};
let freezed_utxos = lock_freezed_utxos()?;
// We filter out the freezed utxos from available list
let available_outpoints: Vec<(OutPoint, OwnedOutput)> = sp_wallet
.get_outputs()
.iter()
.filter_map(|(outpoint, output)| {
if !freezed_utxos.contains(&outpoint) {
Some((*outpoint, output.clone()))
} else {
None
}
})
.collect();
// If we had mandatory inputs, we would make sure to put them at the top of the list
// We don't care for faucet though
// We try to pay the faucet amount
if let Ok(unsigned_transaction) = create_transaction(
available_outpoints,
sp_wallet.get_sp_client(),
vec![recipient],
Some(Vec::from_hex(commitment).unwrap()),
FeeRate::from_sat_per_vb(fee_estimate.to_sat() as f32),
) {
let final_tx = sign_transaction(sp_wallet.get_sp_client(), unsigned_transaction)?;
let partial_tweak = compute_partial_tweak_to_transaction(&final_tx)?;
let daemon = DAEMON
.get()
.ok_or(Error::msg("DAEMON not initialized"))?
.lock_anyhow()?;
// First check that mempool accept it
daemon.test_mempool_accept(&final_tx)?;
let txid = daemon.broadcast(&final_tx)?;
log::debug!("Sent tx {}", txid);
// We immediately add the new tx to our wallet to prevent accidental double spend
check_transaction_alone(sp_wallet, &final_tx, &partial_tweak)?;
Ok((final_tx, partial_tweak))
} else {
// let's try to spend directly from the mining address
let secp = Secp256k1::signing_only();
let keypair = Keypair::new(&secp, &mut thread_rng());
// we first spend from core to the pubkey we just created
let (core_tx, fee_rate) = spend_from_core(keypair.x_only_public_key().0)?;
// check that the first output of the transaction pays to the key we just created
debug_assert!(
core_tx.output[0].script_pubkey
== ScriptBuf::new_p2tr_tweaked(
keypair.x_only_public_key().0.dangerous_assume_tweaked()
)
);
// This is ugly and can be streamlined
// create a new transaction that spends the newly created UTXO to the sp_address
let mut faucet_tx = Transaction {
input: vec![TxIn {
previous_output: OutPoint::new(core_tx.txid(), 0),
..Default::default()
}],
output: vec![],
version: Version::TWO,
lock_time: LockTime::ZERO,
};
// now do the silent payment operations with the final recipient address
let partial_secret = calculate_partial_secret(
&[(keypair.secret_key(), true)],
&[(core_tx.txid().to_string(), 0)],
)?;
let ext_output_key: XOnlyPublicKey =
generate_recipient_pubkeys(vec![sp_address.into()], partial_secret)?
.into_values()
.flatten()
.collect::<Vec<XOnlyPublicKey>>()
.get(0)
.expect("Failed to generate keys")
.to_owned();
let change_sp_address = sp_wallet.get_sp_client().get_receiving_address();
let change_output_key: XOnlyPublicKey =
generate_recipient_pubkeys(vec![change_sp_address], partial_secret)?
.into_values()
.flatten()
.collect::<Vec<XOnlyPublicKey>>()
.get(0)
.expect("Failed to generate keys")
.to_owned();
let ext_spk = ScriptBuf::new_p2tr_tweaked(ext_output_key.dangerous_assume_tweaked());
let change_spk = ScriptBuf::new_p2tr_tweaked(change_output_key.dangerous_assume_tweaked());
let mut op_return = PushBytesBuf::new();
op_return.extend_from_slice(&Vec::from_hex(commitment)?)?;
let data_spk = ScriptBuf::new_op_return(op_return);
// Take some margin to pay for the fees
if core_tx.output[0].value < FAUCET_AMT * 4 {
return Err(Error::msg("Not enough funds"));
}
let change_amt = core_tx.output[0].value.checked_sub(FAUCET_AMT).unwrap();
faucet_tx.output.push(TxOut {
value: FAUCET_AMT,
script_pubkey: ext_spk,
});
faucet_tx.output.push(TxOut {
value: change_amt,
script_pubkey: change_spk,
});
faucet_tx.output.push(TxOut {
value: Amount::from_sat(0),
script_pubkey: data_spk,
});
// dummy signature only used for fee estimation
faucet_tx.input[0].witness.push([1; 64].to_vec());
let abs_fee = fee_rate
.checked_mul(faucet_tx.weight().to_vbytes_ceil())
.ok_or_else(|| Error::msg("Fee rate multiplication overflowed"))?;
// reset the witness to empty
faucet_tx.input[0].witness = Witness::new();
faucet_tx.output[1].value -= abs_fee;
let first_tx_outputs = vec![core_tx.output[0].clone()];
let prevouts = Prevouts::All(&first_tx_outputs);
let hash_ty = TapSighashType::Default;
let mut cache = SighashCache::new(&faucet_tx);
let sighash = cache.taproot_key_spend_signature_hash(0, &prevouts, hash_ty)?;
let msg = Secp256k1Message::from_digest(sighash.into_32());
let sig = secp.sign_schnorr_with_rng(&msg, &keypair, &mut thread_rng());
let final_sig = Signature { sig, hash_ty };
faucet_tx.input[0].witness.push(final_sig.to_vec());
{
let daemon = DAEMON
.get()
.ok_or(Error::msg("DAEMON not initialized"))?
.lock_anyhow()?;
// We don't worry about core_tx being refused by core
daemon.broadcast(&core_tx)?;
daemon.test_mempool_accept(&faucet_tx)?;
let txid = daemon.broadcast(&faucet_tx)?;
log::debug!("Sent tx {}", txid);
}
let partial_tweak = compute_partial_tweak_to_transaction(&faucet_tx)?;
check_transaction_alone(sp_wallet, &faucet_tx, &partial_tweak)?;
Ok((faucet_tx, partial_tweak))
}
}
pub fn handle_faucet_request(msg: &FaucetMessage) -> Result<NewTxMessage> {
let sp_address = SilentPaymentAddress::try_from(msg.sp_address.as_str())?;
log::debug!("Sending bootstrap coins to {}", sp_address);
// send bootstrap coins to this sp_address
let (tx, partial_tweak) = faucet_send(sp_address, &msg.commitment)?;
Ok(NewTxMessage::new(
serialize(&tx).to_lower_hex_string(),
Some(partial_tweak.to_string()),
))
}

79
src/idnot.rs
View File

@ -1,79 +0,0 @@
use reqwest::{Client, Url};
use jwt_simple::prelude::*;
use serde::Deserialize;
use std::error::Error;
use crate::Config;
#[derive(Debug, Clone)]
pub struct IdNotVariables {
pub idnot_client_id: String,
pub idnot_client_secret: String,
pub idnot_redirect_uri: String,
pub idnot_base_url: String,
pub idnot_connexion_url: String,
}
#[derive(Debug, Deserialize)]
pub struct IdNotToken {
pub access_token: String,
pub expires_in: String,
pub id_token: String,
pub token_type: String,
}
#[derive(Debug, Deserialize)]
pub struct IdNotJwtPayload {
pub sub: String,
pub profile_idn: String,
pub entity_idn: String,
}
pub async fn get_id_not_token(
code: &str,
vars: &IdNotVariables,
// ) -> Result<IdNotJwtPayload, Box<dyn Error>> {
) -> Result<(), anyhow::Error> {
// Build the full URL
let url = Url::parse(&format!("{}{}", vars.idnot_base_url, vars.idnot_connexion_url))?;
let params = [
("client_id", vars.idnot_client_id.as_str()),
("client_secret", vars.idnot_client_secret.as_str()),
("redirect_uri", vars.idnot_redirect_uri.as_str()),
("code", &code),
("grant_type", "authorization_code"),
];
let client = Client::new();
let req = client
.post(url.as_str())
.query(&params)
.build()?;
log::debug!("{:#?}", req);
// let resp = client.execute(req).await?;
// log::debug!("{:#?}", resp);
// If status != 200, print the body as error and bail
// if !resp.status().is_success() {
// let err_text = resp.text().await.unwrap_or_else(|_| "<no body>".into());
// log::error!("Error response from {}: {}", url, err_text);
// return Err(anyhow::Error::msg("Non-200 response"));
// }
// Parse JSON into our token struct
// let decoded_token: IdNotToken = resp.json().await?;
// log::debug!("{:?}", decoded_token);
// let token_data: IdNotJwtPayload = // &decoded_token.id_token)
// .map_err(|e| {
// eprintln!("JWT decode error: {}", e);
// "Failed to decode id_token"
// })?;
Ok(())
}

584
src/main.rs
View File

@ -1,584 +0,0 @@
use std::{
collections::{HashMap, HashSet},
env,
fmt::Debug,
fs,
io::{Read, Write},
net::SocketAddr,
path::PathBuf,
str::FromStr,
sync::{Mutex, MutexGuard, OnceLock},
};
use bitcoincore_rpc::{
bitcoin::secp256k1::SecretKey,
json::{self as bitcoin_json},
};
use commit::{lock_members, MEMBERLIST};
use futures_util::{future, pin_mut, stream::TryStreamExt, FutureExt, StreamExt};
use idnot::{get_id_not_token, IdNotVariables};
use log::{debug, error, warn};
use message::{broadcast_message, process_message, BroadcastType, MessageCache, MESSAGECACHE};
use scan::{check_transaction_alone, compute_partial_tweak_to_transaction};
use sdk_common::sp_client::{
bitcoin::secp256k1::rand::{thread_rng, Rng},
bitcoin::OutPoint,
SpClient, SpendKey,
};
use sdk_common::{
error::AnkError,
network::{AnkFlag, NewTxMessage},
};
use sdk_common::{
network::HandshakeMessage,
pcd::Member,
process::{lock_processes, Process, CACHEDPROCESSES},
serialization::{OutPointMemberMap, OutPointProcessMap},
silentpayments::SpWallet,
sp_client::{
bitcoin::{
consensus::deserialize,
hex::{DisplayHex, FromHex},
Amount, Network, Transaction,
},
silentpayments::SilentPaymentAddress,
OwnedOutput,
},
MutexExt,
};
use serde_json::{json, Value};
use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}};
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tokio_tungstenite::tungstenite::Message;
use anyhow::{Error, Result};
use zeromq::{Socket, SocketRecv};
mod commit;
mod config;
mod daemon;
mod electrumclient;
mod faucet;
mod message;
mod scan;
mod commit;
mod idnot;
use crate::config::Config;
use crate::{
daemon::{Daemon, RpcCall},
scan::scan_blocks,
};
type Tx = UnboundedSender<Message>;
type PeerMap = Mutex<HashMap<SocketAddr, Tx>>;
pub(crate) static PEERMAP: OnceLock<PeerMap> = OnceLock::new();
pub(crate) static DAEMON: OnceLock<Mutex<Box<dyn RpcCall>>> = OnceLock::new();
pub static FREEZED_UTXOS: OnceLock<Mutex<HashSet<OutPoint>>> = OnceLock::new();
pub fn lock_freezed_utxos() -> Result<MutexGuard<'static, HashSet<OutPoint>>, Error> {
FREEZED_UTXOS
.get_or_init(|| Mutex::new(HashSet::new()))
.lock_anyhow()
}
#[derive(Debug)]
pub struct StateFile {
path: PathBuf,
}
impl StateFile {
fn new(path: PathBuf) -> Self {
Self { path }
}
fn create(&self) -> Result<()> {
let parent: PathBuf;
if let Some(dir) = self.path.parent() {
if !dir.ends_with(".4nk") {
return Err(Error::msg("parent dir must be \".4nk\""));
}
parent = dir.to_path_buf();
} else {
return Err(Error::msg("wallet file has no parent dir"));
}
// Ensure the parent directory exists
if !parent.exists() {
fs::create_dir_all(parent)?;
}
// Create a new file
fs::File::create(&self.path)?;
Ok(())
}
fn save(&self, json: &Value) -> Result<()> {
let mut f = fs::File::options()
.write(true)
.truncate(true)
.open(&self.path)?;
let stringified = serde_json::to_string(&json)?;
let bin = stringified.as_bytes();
f.write_all(bin)?;
Ok(())
}
fn load(&self) -> Result<Value> {
let mut f = fs::File::open(&self.path)?;
let mut content = vec![];
f.read_to_end(&mut content)?;
let res: Value = serde_json::from_slice(&content)?;
Ok(res)
}
}
#[derive(Debug)]
pub struct DiskStorage {
pub wallet_file: StateFile,
pub processes_file: StateFile,
pub members_file: StateFile,
}
pub static STORAGE: OnceLock<Mutex<DiskStorage>> = OnceLock::new();
const FAUCET_AMT: Amount = Amount::from_sat(10_000);
pub(crate) static WALLET: OnceLock<Mutex<SpWallet>> = OnceLock::new();
fn handle_new_tx_request(new_tx_msg: &NewTxMessage) -> Result<()> {
let tx = deserialize::<Transaction>(&Vec::from_hex(&new_tx_msg.transaction)?)?;
let daemon = DAEMON.get().unwrap().lock_anyhow()?;
daemon.test_mempool_accept(&tx)?;
daemon.broadcast(&tx)?;
Ok(())
}
async fn handle_http_connection(mut raw_stream: TcpStream, addr: SocketAddr, vars: IdNotVariables) {
debug!("Incoming HTTP connection from: {}", addr);
// 1) Read up to 4KiB for the request headers
let mut buf = vec![0u8; 4096];
let n = match raw_stream.read(&mut buf).await {
Ok(0) => return, // client closed connection
Ok(n) => n,
Err(err) => {
debug!("Failed to read from socket: {}", err);
return;
}
};
// 2) Convert to &str
let req = match std::str::from_utf8(&buf[..n]) {
Ok(s) => s,
Err(err) => {
debug!("Invalid UTF-8 request: {}", err);
return;
}
};
// 3) Parse request-line: "POST /api/v1/idnot/user?code=XYZ HTTP/1.1"
let mut lines = req.split("\r\n");
let request_line = match lines.next() {
Some(l) => l,
None => return,
};
let mut parts = request_line.split_whitespace();
let method = parts.next().unwrap_or("");
let uri = parts.next().unwrap_or("");
// 4) Only handle POST /api/v1/idnot/user
if method == "POST" && uri.starts_with("/api/v1/idnot/user/") {
let code_opt = uri
.split('/')
.last()
.filter(|&code| !code.is_empty());
if let Some(code) = code_opt {
if let Ok(()) = get_id_not_token(code, &vars).await {
let response =
"HTTP/1.1 200 OK\r\n\
Content-Type: application/json\r\n\
Content-Length: 0\r\n\r\n";
let _ = raw_stream.write_all(response.as_bytes()).await;
} else {
log::error!("Failed to get token");
}
return;
} else {
// Missing code → 400 Bad Request
let body = json!({ "error": "code is required" }).to_string();
let response = format!(
"HTTP/1.1 400 Bad Request\r\n\
Content-Type: application/json\r\n\
Content-Length: {}\r\n\r\n\
{}",
body.len(),
body
);
let _ = raw_stream.write_all(response.as_bytes()).await;
return;
}
}
// 5) Fallback: 404 Not Found
let response = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n";
let _ = raw_stream.write_all(response.as_bytes()).await;
}
async fn handle_ws_connection(raw_stream: TcpStream, addr: SocketAddr, our_sp_address: SilentPaymentAddress) {
debug!("Incoming TCP connection from: {}", addr);
let peers = PEERMAP.get().expect("Peer Map not initialized");
let ws_stream = match tokio_tungstenite::accept_async(raw_stream).await {
Ok(stream) => {
debug!("WebSocket connection established");
stream
}
Err(e) => {
log::error!("WebSocket handshake failed for {}: {}", addr, e);
return;
}
};
// Insert the write part of this peer to the peer map.
let (tx, rx) = unbounded_channel();
match peers.lock_anyhow() {
Ok(mut peer_map) => peer_map.insert(addr, tx),
Err(e) => {
log::error!("{}", e);
panic!();
}
};
let processes = lock_processes().unwrap().clone();
let members = lock_members().unwrap().clone();
let init_msg = HandshakeMessage::new(
our_sp_address.to_string(),
OutPointMemberMap(members),
OutPointProcessMap(processes),
);
if let Err(e) = broadcast_message(
AnkFlag::Handshake,
format!("{}", init_msg.to_string()),
BroadcastType::Sender(addr),
) {
log::error!("Failed to send init message: {}", e);
return;
}
let (outgoing, incoming) = ws_stream.split();
let broadcast_incoming = incoming.try_for_each(|msg| {
if let Ok(raw_msg) = msg.to_text() {
// debug!("Received msg: {}", raw_msg);
process_message(raw_msg, addr);
} else {
debug!("Received non-text message {} from peer {}", msg, addr);
}
future::ok(())
});
let receive_from_others = UnboundedReceiverStream::new(rx)
.map(Ok)
.forward(outgoing)
.map(|result| {
if let Err(e) = result {
debug!("Error sending message: {}", e);
}
});
pin_mut!(broadcast_incoming, receive_from_others);
future::select(broadcast_incoming, receive_from_others).await;
debug!("{} disconnected", &addr);
peers.lock().unwrap().remove(&addr);
}
fn create_new_tx_message(transaction: Vec<u8>) -> Result<NewTxMessage> {
// debug!("Creating tx message");
let tx: Transaction = deserialize(&transaction)?;
if tx.is_coinbase() {
return Err(Error::msg("Can't process coinbase transaction"));
}
let partial_tweak = compute_partial_tweak_to_transaction(&tx)?;
let sp_wallet = WALLET
.get()
.ok_or_else(|| Error::msg("Wallet not initialized"))?
.lock_anyhow()?;
check_transaction_alone(sp_wallet, &tx, &partial_tweak)?;
Ok(NewTxMessage::new(
transaction.to_lower_hex_string(),
Some(partial_tweak.to_string()),
))
}
async fn handle_zmq(zmq_url: String, electrum_url: String) {
debug!("Starting listening on Core");
let mut socket = zeromq::SubSocket::new();
socket.connect(&zmq_url).await.unwrap();
socket.subscribe("rawtx").await.unwrap();
socket.subscribe("hashblock").await.unwrap();
loop {
let core_msg = match socket.recv().await {
Ok(m) => m,
Err(e) => {
error!("Zmq error: {}", e);
continue;
}
};
debug!("Received a message");
let payload: String = if let (Some(topic), Some(data)) = (core_msg.get(0), core_msg.get(1))
{
debug!("topic: {}", std::str::from_utf8(&topic).unwrap());
match std::str::from_utf8(&topic) {
Ok("rawtx") => match create_new_tx_message(data.to_vec()) {
Ok(m) => {
debug!("Created message");
serde_json::to_string(&m).expect("This shouldn't fail")
}
Err(e) => {
error!("{}", e);
continue;
}
},
Ok("hashblock") => match scan_blocks(0, &electrum_url) {
Ok(_) => continue,
Err(e) => {
error!("{}", e);
continue;
}
},
_ => {
error!("Unexpected message in zmq");
continue;
}
}
} else {
error!("Empty message");
continue;
};
if let Err(e) = broadcast_message(AnkFlag::NewTx, payload, BroadcastType::ToAll) {
log::error!("{}", e.to_string());
}
}
}
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
env_logger::init();
// todo: take the path to conf file as argument
// default to "./.conf"
let config = Config::read_from_file(".conf")?;
if config.network == Network::Bitcoin {
warn!("Running on mainnet, you're on your own");
}
MESSAGECACHE
.set(MessageCache::new())
.expect("Message Cache initialization failed");
PEERMAP
.set(PeerMap::new(HashMap::new()))
.expect("PeerMap initialization failed");
// Connect the rpc daemon
DAEMON
.set(Mutex::new(Box::new(Daemon::connect(
config.core_wallet,
config.core_url,
config.network,
)?)))
.expect("DAEMON initialization failed");
let current_tip: u32 = DAEMON
.get()
.unwrap()
.lock_anyhow()?
.get_current_height()?
.try_into()?;
let mut app_dir = PathBuf::from_str(&env::var("HOME")?)?;
app_dir.push(config.data_dir);
let mut wallet_file = app_dir.clone();
wallet_file.push(&config.wallet_name);
let mut processes_file = app_dir.clone();
processes_file.push("processes");
let mut members_file = app_dir.clone();
members_file.push("members");
let wallet_file = StateFile::new(wallet_file);
let processes_file = StateFile::new(processes_file);
let members_file = StateFile::new(members_file);
// load an existing sp_wallet, or create a new one
let sp_wallet: SpWallet = match wallet_file.load() {
Ok(wallet) => {
// TODO: Verify the wallet is compatible with the current network
serde_json::from_value(wallet)?
}
Err(_) => {
// Create a new wallet file if it doesn't exist or fails to load
wallet_file.create()?;
let mut rng = thread_rng();
let new_client = SpClient::new(
SecretKey::new(&mut rng),
SpendKey::Secret(SecretKey::new(&mut rng)),
config.network,
)
.expect("Failed to create a new SpClient");
let mut sp_wallet = SpWallet::new(new_client);
// Set birthday and update scan information
sp_wallet.set_birthday(current_tip);
sp_wallet.set_last_scan(current_tip);
// Save the newly created wallet to disk
let json = serde_json::to_value(sp_wallet.clone())?;
wallet_file.save(&json)?;
sp_wallet
}
};
let cached_processes: HashMap<OutPoint, Process> = match processes_file.load() {
Ok(processes) => {
let deserialized: OutPointProcessMap = serde_json::from_value(processes)?;
deserialized.0
}
Err(_) => {
debug!("creating process file at {}", processes_file.path.display());
processes_file.create()?;
HashMap::new()
}
};
let members: HashMap<OutPoint, Member> = match members_file.load() {
Ok(members) => {
let deserialized: OutPointMemberMap = serde_json::from_value(members)?;
deserialized.0
}
Err(_) => {
debug!("creating members file at {}", members_file.path.display());
members_file.create()?;
HashMap::new()
}
};
{
let utxo_to_freeze: HashSet<OutPoint> = cached_processes
.iter()
.map(|(_, process)| process.get_last_unspent_outpoint().unwrap())
.collect();
let mut freezed_utxos = lock_freezed_utxos()?;
*freezed_utxos = utxo_to_freeze;
}
let our_sp_address = sp_wallet.get_sp_client().get_receiving_address();
log::info!("Using wallet with address {}", our_sp_address,);
log::info!(
"Found {} outputs for a total balance of {}",
sp_wallet.get_outputs().len(),
sp_wallet.get_balance()
);
let last_scan = sp_wallet.get_last_scan();
WALLET
.set(Mutex::new(sp_wallet))
.expect("Failed to initialize WALLET");
CACHEDPROCESSES
.set(Mutex::new(cached_processes))
.expect("Failed to initialize CACHEDPROCESSES");
MEMBERLIST
.set(Mutex::new(members))
.expect("Failed to initialize MEMBERLIST");
let storage = DiskStorage {
wallet_file,
processes_file,
members_file,
};
STORAGE.set(Mutex::new(storage)).unwrap();
if last_scan < current_tip {
log::info!("Scanning for our outputs");
scan_blocks(current_tip - last_scan, &config.electrum_url)?;
}
// Subscribe to Bitcoin Core
tokio::spawn(handle_zmq(config.zmq_url, config.electrum_url));
// Create the TcpListener for http request
let try_socket = TcpListener::bind(config.http_url).await;
let http_listener = try_socket.expect("Failed to bind");
if config.idnot_client_id.is_some() && config.idnot_client_secret.is_some() && config.idnot_redirect_uri.is_some() {
let id_not_vars = IdNotVariables {
idnot_base_url: "https://qual-connexion.idnot.fr".to_owned(),
idnot_client_id: config.idnot_client_id.unwrap(),
idnot_client_secret: config.idnot_client_secret.unwrap(),
idnot_connexion_url: "/IdPOAuth2/token/idnot_idp_v1".to_owned(),
idnot_redirect_uri: config.idnot_redirect_uri.unwrap(),
};
log::debug!("{:?}", id_not_vars);
// Spawn a task to handle HTTP connections
tokio::spawn(async move {
while let Ok((stream, addr)) = http_listener.accept().await {
tokio::spawn(handle_http_connection(stream, addr, id_not_vars.clone()));
}
});
}
// Create the event loop and TCP listener we'll accept connections on.
let try_socket = TcpListener::bind(config.ws_url).await;
let ws_listener = try_socket.expect("Failed to bind");
tokio::spawn(MessageCache::clean_up());
// Let's spawn the handling of each connection in a separate task.
while let Ok((stream, addr)) = ws_listener.accept().await {
tokio::spawn(handle_ws_connection(stream, addr, our_sp_address));
}
Ok(())
}

233
src/message.rs
View File

@ -1,233 +0,0 @@
use anyhow::{Error, Result};
use std::{
collections::HashMap,
net::SocketAddr,
sync::{Mutex, OnceLock},
time::{Duration, Instant},
};
use tokio::time;
use tokio_tungstenite::tungstenite::Message;
use sdk_common::network::{AnkFlag, CommitMessage, Envelope, FaucetMessage, NewTxMessage};
use crate::{
commit::handle_commit_request, faucet::handle_faucet_request, handle_new_tx_request, PEERMAP,
};
pub(crate) static MESSAGECACHE: OnceLock<MessageCache> = OnceLock::new();
const MESSAGECACHEDURATION: Duration = Duration::from_secs(20);
const MESSAGECACHEINTERVAL: Duration = Duration::from_secs(5);
#[derive(Debug)]
pub(crate) struct MessageCache {
store: Mutex<HashMap<String, Instant>>,
}
impl MessageCache {
pub fn new() -> Self {
Self {
store: Mutex::new(HashMap::new()),
}
}
fn insert(&self, key: String) {
let mut store = self.store.lock().unwrap();
store.insert(key, Instant::now());
}
fn remove(&self, key: &str) {
let mut store = self.store.lock().unwrap();
store.remove(key);
}
fn contains(&self, key: &str) -> bool {
let store = self.store.lock().unwrap();
store.contains_key(key)
}
pub async fn clean_up() {
let cache = MESSAGECACHE.get().unwrap();
let mut interval = time::interval(MESSAGECACHEINTERVAL);
loop {
interval.tick().await;
let mut store = cache.store.lock().unwrap();
let now = Instant::now();
store.retain(|_, entrytime| now.duration_since(*entrytime) <= MESSAGECACHEDURATION);
}
}
}
pub(crate) enum BroadcastType {
Sender(SocketAddr),
#[allow(dead_code)]
ExcludeSender(SocketAddr),
#[allow(dead_code)]
ToAll,
}
pub(crate) fn broadcast_message(
flag: AnkFlag,
payload: String,
broadcast: BroadcastType,
) -> Result<()> {
let peers = PEERMAP.get().ok_or(Error::msg("Unitialized peer map"))?;
let ank_msg = Envelope {
flag,
content: payload,
};
let msg = Message::Text(serde_json::to_string(&ank_msg)?);
match ank_msg.flag {
AnkFlag::Cipher => log::debug!("Broadcasting cipher"),
AnkFlag::Handshake => log::debug!("Broadcasting handshake"),
_ => log::debug!("Broadcasting {} message: {}", ank_msg.flag.as_str(), msg),
}
match broadcast {
BroadcastType::Sender(addr) => {
peers
.lock()
.map_err(|e| Error::msg(format!("Failed to lock peers: {}", e.to_string())))?
.iter()
.find(|(peer_addr, _)| peer_addr == &&addr)
.ok_or(Error::msg("Failed to find the sender in the peer_map"))?
.1
.send(msg)?;
}
BroadcastType::ExcludeSender(addr) => {
peers
.lock()
.map_err(|e| Error::msg(format!("Failed to lock peers: {}", e.to_string())))?
.iter()
.filter(|(peer_addr, _)| peer_addr != &&addr)
.for_each(|(_, peer_tx)| {
let _ = peer_tx.send(msg.clone());
});
}
BroadcastType::ToAll => {
peers
.lock()
.map_err(|e| Error::msg(format!("Failed to lock peers: {}", e.to_string())))?
.iter()
.for_each(|(_, peer_tx)| {
let _ = peer_tx.send(msg.clone());
});
}
}
Ok(())
}
fn process_faucet_message(ank_msg: Envelope, addr: SocketAddr) {
log::debug!("Received a faucet message");
if let Ok(mut content) = serde_json::from_str::<FaucetMessage>(&ank_msg.content) {
match handle_faucet_request(&content) {
Ok(new_tx_msg) => {
log::debug!(
"Obtained new_tx_msg: {}",
serde_json::to_string(&new_tx_msg).unwrap()
);
}
Err(e) => {
log::error!("Failed to send faucet tx: {}", e);
content.error = Some(e.into());
let payload = serde_json::to_string(&content).expect("Message type shouldn't fail");
if let Err(e) =
broadcast_message(AnkFlag::Faucet, payload, BroadcastType::Sender(addr))
{
log::error!("Failed to broadcast message: {}", e);
}
}
}
} else {
log::error!("Invalid content for faucet message");
}
}
fn process_new_tx_message(ank_msg: Envelope, addr: SocketAddr) {
log::debug!("Received a new tx message");
if let Ok(mut new_tx_msg) = serde_json::from_str::<NewTxMessage>(&ank_msg.content) {
if let Err(e) = handle_new_tx_request(&mut new_tx_msg) {
log::error!("handle_new_tx_request returned error: {}", e);
new_tx_msg.error = Some(e.into());
if let Err(e) = broadcast_message(
AnkFlag::NewTx,
serde_json::to_string(&new_tx_msg).expect("This shouldn't fail"),
BroadcastType::Sender(addr),
) {
log::error!("Failed to broadcast message: {}", e);
}
}
} else {
log::error!("Invalid content for new_tx message");
}
}
fn process_cipher_message(ank_msg: Envelope, addr: SocketAddr) {
// For now we just send it to everyone
log::debug!("Received a cipher message");
if let Err(e) = broadcast_message(
AnkFlag::Cipher,
ank_msg.content,
BroadcastType::ExcludeSender(addr),
) {
log::error!("Failed to send message with error: {}", e);
}
}
fn process_commit_message(ank_msg: Envelope, addr: SocketAddr) {
if let Ok(mut commit_msg) = serde_json::from_str::<CommitMessage>(&ank_msg.content) {
match handle_commit_request(commit_msg.clone()) {
Ok(new_outpoint) => log::debug!("Processed commit msg for outpoint {}", new_outpoint),
Err(e) => {
log::error!("handle_commit_request returned error: {}", e);
// Temporary fix: we remove the message from the cache in case the client wants to try again
let cache = MESSAGECACHE.get().expect("Cache should be initialized");
cache.remove(ank_msg.to_string().as_str());
commit_msg.error = Some(e.into());
if let Err(e) = broadcast_message(
AnkFlag::Commit,
serde_json::to_string(&commit_msg).expect("This shouldn't fail"),
BroadcastType::Sender(addr),
) {
log::error!("Failed to broadcast message: {}", e);
}
}
};
}
}
fn process_unknown_message(ank_msg: Envelope, addr: SocketAddr) {
log::debug!("Received an unknown message");
if let Err(e) = broadcast_message(
AnkFlag::Unknown,
ank_msg.content,
BroadcastType::ExcludeSender(addr),
) {
log::error!("Failed to send message with error: {}", e);
}
}
pub fn process_message(raw_msg: &str, addr: SocketAddr) {
let cache = MESSAGECACHE.get().expect("Cache should be initialized");
if cache.contains(raw_msg) {
log::debug!("Message already processed, dropping");
return;
} else {
cache.insert(raw_msg.to_owned());
}
match serde_json::from_str::<Envelope>(raw_msg) {
Ok(ank_msg) => match ank_msg.flag {
AnkFlag::Faucet => process_faucet_message(ank_msg, addr),
AnkFlag::NewTx => process_new_tx_message(ank_msg, addr),
AnkFlag::Cipher => process_cipher_message(ank_msg, addr),
AnkFlag::Commit => process_commit_message(ank_msg, addr),
AnkFlag::Unknown => process_unknown_message(ank_msg, addr),
AnkFlag::Handshake => log::debug!("Received init message from {}", addr),
},
Err(_) => log::error!("Failed to parse network message"),
}
}

362
src/scan.rs
View File

@ -1,362 +0,0 @@
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::MutexGuard;
use anyhow::{Error, Result};
use bitcoincore_rpc::bitcoin::absolute::Height;
use electrum_client::ElectrumApi;
use sdk_common::silentpayments::SpWallet;
use sdk_common::sp_client::bitcoin::bip158::BlockFilter;
use sdk_common::sp_client::bitcoin::hex::DisplayHex;
use sdk_common::sp_client::bitcoin::secp256k1::{All, PublicKey, Scalar, Secp256k1, SecretKey};
use sdk_common::sp_client::bitcoin::{BlockHash, OutPoint, Transaction, TxOut, XOnlyPublicKey};
use sdk_common::sp_client::silentpayments::receiving::Receiver;
use sdk_common::sp_client::silentpayments::utils::receiving::{
calculate_tweak_data, get_pubkey_from_input,
};
use sdk_common::sp_client::{OutputSpendStatus, OwnedOutput};
use tokio::time::Instant;
use crate::{electrumclient, MutexExt, DAEMON, STORAGE, WALLET};
pub fn compute_partial_tweak_to_transaction(tx: &Transaction) -> Result<PublicKey> {
let daemon = DAEMON.get().ok_or(Error::msg("DAEMON not initialized"))?;
let mut outpoints: Vec<(String, u32)> = Vec::with_capacity(tx.input.len());
let mut pubkeys: Vec<PublicKey> = Vec::with_capacity(tx.input.len());
// TODO we should cache transactions to prevent multiple rpc request when transaction spends multiple outputs from the same tx
for input in tx.input.iter() {
outpoints.push((
input.previous_output.txid.to_string(),
input.previous_output.vout,
));
let prev_tx = daemon
.lock_anyhow()?
.get_transaction(&input.previous_output.txid, None)
.map_err(|e| Error::msg(format!("Failed to find previous transaction: {}", e)))?;
if let Some(output) = prev_tx.output.get(input.previous_output.vout as usize) {
match get_pubkey_from_input(
&input.script_sig.to_bytes(),
&input.witness.to_vec(),
&output.script_pubkey.to_bytes(),
) {
Ok(Some(pubkey)) => pubkeys.push(pubkey),
Ok(None) => continue,
Err(e) => {
return Err(Error::msg(format!(
"Can't extract pubkey from input: {}",
e
)))
}
}
} else {
return Err(Error::msg("Transaction with a non-existing input"));
}
}
let input_pub_keys: Vec<&PublicKey> = pubkeys.iter().collect();
let partial_tweak = calculate_tweak_data(&input_pub_keys, &outpoints)?;
Ok(partial_tweak)
}
fn get_script_to_secret_map(
sp_receiver: &Receiver,
tweak_data_vec: Vec<String>,
scan_key_scalar: Scalar,
secp: &Secp256k1<All>,
) -> Result<HashMap<[u8; 34], PublicKey>> {
let mut res = HashMap::new();
let shared_secrets: Result<Vec<PublicKey>> = tweak_data_vec
.into_iter()
.map(|s| {
let x = PublicKey::from_str(&s).map_err(Error::new)?;
x.mul_tweak(secp, &scan_key_scalar).map_err(Error::new)
})
.collect();
let shared_secrets = shared_secrets?;
for shared_secret in shared_secrets {
let spks = sp_receiver.get_spks_from_shared_secret(&shared_secret)?;
for spk in spks.into_values() {
res.insert(spk, shared_secret);
}
}
Ok(res)
}
pub fn check_transaction_alone(
mut wallet: MutexGuard<SpWallet>,
tx: &Transaction,
tweak_data: &PublicKey,
) -> Result<HashMap<OutPoint, OwnedOutput>> {
let updates = match wallet.update_with_transaction(tx, tweak_data, 0) {
Ok(updates) => updates,
Err(e) => {
log::debug!("Error while checking transaction: {}", e);
HashMap::new()
}
};
if updates.len() > 0 {
let storage = STORAGE
.get()
.ok_or_else(|| Error::msg("Failed to get STORAGE"))?;
storage
.lock_anyhow()?
.wallet_file
.save(&serde_json::to_value(wallet.clone())?)?;
}
Ok(updates)
}
fn check_block(
blkfilter: BlockFilter,
blkhash: BlockHash,
candidate_spks: Vec<&[u8; 34]>,
owned_spks: Vec<Vec<u8>>,
) -> Result<bool> {
// check output scripts
let mut scripts_to_match: Vec<_> = candidate_spks.into_iter().map(|spk| spk.as_ref()).collect();
// check input scripts
scripts_to_match.extend(owned_spks.iter().map(|spk| spk.as_slice()));
// note: match will always return true for an empty query!
if !scripts_to_match.is_empty() {
Ok(blkfilter.match_any(&blkhash, &mut scripts_to_match.into_iter())?)
} else {
Ok(false)
}
}
fn scan_block_outputs(
sp_receiver: &Receiver,
txdata: &Vec<Transaction>,
blkheight: u64,
spk2secret: HashMap<[u8; 34], PublicKey>,
) -> Result<HashMap<OutPoint, OwnedOutput>> {
let mut res: HashMap<OutPoint, OwnedOutput> = HashMap::new();
// loop over outputs
for tx in txdata {
let txid = tx.txid();
// collect all taproot outputs from transaction
let p2tr_outs: Vec<(usize, &TxOut)> = tx
.output
.iter()
.enumerate()
.filter(|(_, o)| o.script_pubkey.is_p2tr())
.collect();
if p2tr_outs.is_empty() {
continue;
}; // no taproot output
let mut secret: Option<PublicKey> = None;
// Does this transaction contains one of the outputs we already found?
for spk in p2tr_outs.iter().map(|(_, o)| &o.script_pubkey) {
if let Some(s) = spk2secret.get(spk.as_bytes()) {
// we might have at least one output in this transaction
secret = Some(*s);
break;
}
}
if secret.is_none() {
continue;
}; // we don't have a secret that matches any of the keys
// Now we can just run sp_receiver on all the p2tr outputs
let xonlykeys: Result<Vec<XOnlyPublicKey>> = p2tr_outs
.iter()
.map(|(_, o)| {
XOnlyPublicKey::from_slice(&o.script_pubkey.as_bytes()[2..]).map_err(Error::new)
})
.collect();
let ours = sp_receiver.scan_transaction(&secret.unwrap(), xonlykeys?)?;
let height = Height::from_consensus(blkheight as u32)?;
for (label, map) in ours {
res.extend(p2tr_outs.iter().filter_map(|(i, o)| {
match XOnlyPublicKey::from_slice(&o.script_pubkey.as_bytes()[2..]) {
Ok(key) => {
if let Some(scalar) = map.get(&key) {
match SecretKey::from_slice(&scalar.to_be_bytes()) {
Ok(tweak) => {
let outpoint = OutPoint {
txid,
vout: *i as u32,
};
let label_str: Option<String>;
if let Some(l) = &label {
label_str =
Some(l.as_inner().to_be_bytes().to_lower_hex_string());
} else {
label_str = None;
}
return Some((
outpoint,
OwnedOutput {
blockheight: height,
tweak: tweak.secret_bytes(),
amount: o.value,
script: o.script_pubkey.clone(),
label: label_str,
spend_status: OutputSpendStatus::Unspent,
},
));
}
Err(_) => {
return None;
}
}
}
None
}
Err(_) => None,
}
}));
}
}
Ok(res)
}
fn scan_block_inputs(
our_outputs: &HashMap<OutPoint, OwnedOutput>,
txdata: Vec<Transaction>,
) -> Result<Vec<OutPoint>> {
let mut found = vec![];
for tx in txdata {
for input in tx.input {
let prevout = input.previous_output;
if our_outputs.contains_key(&prevout) {
found.push(prevout);
}
}
}
Ok(found)
}
pub fn scan_blocks(mut n_blocks_to_scan: u32, electrum_url: &str) -> anyhow::Result<()> {
log::info!("Starting a rescan");
let electrum_client = electrumclient::create_electrum_client(electrum_url)?;
let mut sp_wallet = WALLET
.get()
.ok_or(Error::msg("Wallet not initialized"))?
.lock_anyhow()?;
let core = DAEMON
.get()
.ok_or(Error::msg("DAEMON not initialized"))?
.lock_anyhow()?;
let secp = Secp256k1::new();
let scan_height = sp_wallet.get_last_scan();
let tip_height: u32 = core.get_current_height()?.try_into()?;
// 0 means scan to tip
if n_blocks_to_scan == 0 {
n_blocks_to_scan = tip_height - scan_height;
}
let start = scan_height + 1;
let end = if scan_height + n_blocks_to_scan <= tip_height {
scan_height + n_blocks_to_scan
} else {
tip_height
};
if start > end {
return Ok(());
}
log::info!("start: {} end: {}", start, end);
let mut filters: Vec<(u32, BlockHash, BlockFilter)> = vec![];
for blkheight in start..=end {
filters.push(core.get_filters(blkheight)?);
}
let mut tweak_data_map = electrum_client.sp_tweaks(start as usize)?;
let scan_sk = sp_wallet.get_sp_client().get_scan_key();
let start_time = Instant::now();
for (blkheight, blkhash, blkfilter) in filters {
let spk2secret = match tweak_data_map.remove(&blkheight) {
Some(tweak_data_vec) => get_script_to_secret_map(
&sp_wallet.get_sp_client().sp_receiver,
tweak_data_vec,
scan_sk.into(),
&secp,
)?,
None => HashMap::new(),
};
// check if new possible outputs are payments to us
let candidate_spks: Vec<&[u8; 34]> = spk2secret.keys().collect();
// check if owned inputs are spent
let owned_spks: Vec<Vec<u8>> = sp_wallet
.get_outputs()
.iter()
.map(|(_, output)| {
let script = output.script.to_bytes();
script
})
.collect();
let matched = check_block(blkfilter, blkhash, candidate_spks, owned_spks)?;
if matched {
let blk = core.get_block(blkhash)?;
// scan block for new outputs, and add them to our list
let utxo_created_in_block = scan_block_outputs(
&sp_wallet.get_sp_client().sp_receiver,
&blk.txdata,
blkheight.into(),
spk2secret,
)?;
if !utxo_created_in_block.is_empty() {
sp_wallet.get_mut_outputs().extend(utxo_created_in_block);
}
// update the list of outputs just in case
// utxos may be created and destroyed in the same block
// search inputs and mark as mined
let utxo_destroyed_in_block = scan_block_inputs(sp_wallet.get_outputs(), blk.txdata)?;
if !utxo_destroyed_in_block.is_empty() {
let outputs = sp_wallet.get_mut_outputs();
for outpoint in utxo_destroyed_in_block {
if let Some(output) = outputs.get_mut(&outpoint) {
output.spend_status = OutputSpendStatus::Mined(blkhash.to_string());
}
}
}
}
}
// time elapsed for the scan
log::info!(
"Scan complete in {} seconds",
start_time.elapsed().as_secs()
);
// update last_scan height
sp_wallet.set_last_scan(end);
STORAGE
.get()
.unwrap()
.lock_anyhow()?
.wallet_file
.save(&serde_json::to_value(sp_wallet.clone())?)?;
Ok(())
}