stratum.rs (15245B)
1 use std::{ 2 collections::BTreeMap, 3 net::SocketAddr, 4 sync::{ 5 Arc, 6 atomic::{AtomicU64, Ordering}, 7 }, 8 }; 9 10 use anyhow::{Context, Result, bail}; 11 use serde_json::{Value, json}; 12 use tokio::{ 13 io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, 14 net::{TcpListener, TcpStream, tcp::OwnedWriteHalf}, 15 sync::Mutex, 16 }; 17 18 use crate::{ 19 adapters::p2p::GossipNetwork, 20 app::{ExternalMineJob, SharedNode, debug_logging_enabled}, 21 domain::{STRATUM_EXTRANONCE1_HEX, STRATUM_EXTRANONCE2_SIZE, StratumMineShare}, 22 }; 23 24 #[derive(Clone)] 25 pub struct StratumServer { 26 node: SharedNode, 27 gossip: GossipNetwork, 28 listen_addr: SocketAddr, 29 next_job_salt: Arc<AtomicU64>, 30 } 31 32 #[derive(Clone, Debug)] 33 struct StratumJob { 34 mine: ExternalMineJob, 35 } 36 37 impl StratumServer { 38 pub async fn start( 39 node: SharedNode, 40 gossip: GossipNetwork, 41 listen_addr: SocketAddr, 42 ) -> Result<Self> { 43 let listener = TcpListener::bind(listen_addr) 44 .await 45 .with_context(|| format!("failed to bind Stratum listener on {listen_addr}"))?; 46 let local_addr = listener.local_addr()?; 47 let server = Self { 48 node, 49 gossip, 50 listen_addr: local_addr, 51 next_job_salt: Arc::new(AtomicU64::new(1)), 52 }; 53 tokio::spawn(run_listener(server.clone(), listener)); 54 Ok(server) 55 } 56 57 pub fn listen_addr(&self) -> SocketAddr { 58 self.listen_addr 59 } 60 } 61 62 async fn run_listener(server: StratumServer, listener: TcpListener) { 63 loop { 64 match listener.accept().await { 65 Ok((stream, remote)) => { 66 let server = server.clone(); 67 tokio::spawn(async move { 68 if let Err(error) = handle_connection(server, stream).await { 69 if debug_logging_enabled() { 70 eprintln!("stratum session with {remote} failed: {error:#}"); 71 } 72 } 73 }); 74 } 75 Err(error) if debug_logging_enabled() => { 76 eprintln!("stratum accept failed: {error:#}"); 77 } 78 Err(_) => {} 79 } 80 } 81 } 82 83 async fn handle_connection(server: StratumServer, stream: TcpStream) -> Result<()> { 84 let (read, write) = stream.into_split(); 85 let mut session = StratumSession { 86 server, 87 writer: Arc::new(Mutex::new(write)), 88 authorized_worker: None, 89 jobs: BTreeMap::new(), 90 next_job_id: 1, 91 }; 92 let mut lines = BufReader::new(read).lines(); 93 while let Some(line) = lines.next_line().await? { 94 if line.trim().is_empty() { 95 continue; 96 } 97 let request: Value = serde_json::from_str(&line).context("invalid Stratum JSON")?; 98 session.handle_request(request).await?; 99 } 100 Ok(()) 101 } 102 103 struct StratumSession { 104 server: StratumServer, 105 writer: Arc<Mutex<OwnedWriteHalf>>, 106 authorized_worker: Option<String>, 107 jobs: BTreeMap<String, StratumJob>, 108 next_job_id: u64, 109 } 110 111 impl StratumSession { 112 async fn handle_request(&mut self, request: Value) -> Result<()> { 113 let id = request.get("id").cloned().unwrap_or(Value::Null); 114 let method = request 115 .get("method") 116 .and_then(Value::as_str) 117 .context("Stratum request is missing method")?; 118 match method { 119 "mining.subscribe" => { 120 self.send_response( 121 id, 122 json!([ 123 [["mining.set_difficulty", "iuna"], ["mining.notify", "iuna"]], 124 STRATUM_EXTRANONCE1_HEX, 125 STRATUM_EXTRANONCE2_SIZE 126 ]), 127 ) 128 .await?; 129 } 130 "mining.authorize" => { 131 let worker = request 132 .get("params") 133 .and_then(Value::as_array) 134 .and_then(|params| params.first()) 135 .and_then(Value::as_str) 136 .context("mining.authorize requires worker address")? 137 .to_string(); 138 self.authorized_worker = Some(worker.clone()); 139 self.send_response(id, json!(true)).await?; 140 self.send_job(&worker, true).await?; 141 } 142 "mining.submit" => { 143 let accepted = self.handle_submit(&request).await; 144 match accepted { 145 Ok(true) => self.send_response(id, json!(true)).await?, 146 Ok(false) => { 147 self.send_error(id, 23, "duplicate share or transaction") 148 .await?; 149 } 150 Err(error) => self.send_error(id, 23, &format!("{error:#}")).await?, 151 } 152 } 153 "mining.configure" => { 154 self.send_response(id, json!({})).await?; 155 } 156 "mining.extranonce.subscribe" => { 157 self.send_response(id, json!(true)).await?; 158 } 159 _ => { 160 self.send_error(id, 20, &format!("unsupported method {method}")) 161 .await?; 162 } 163 } 164 Ok(()) 165 } 166 167 async fn send_job(&mut self, worker: &str, clean_jobs: bool) -> Result<()> { 168 let job_id = self.next_job_id.to_string(); 169 self.next_job_id = self.next_job_id.saturating_add(1); 170 let salt = self.server.next_job_salt.fetch_add(1, Ordering::Relaxed); 171 let mine = self 172 .server 173 .node 174 .lock() 175 .await 176 .external_mine_job(recipient_from_worker(worker), salt)?; 177 let difficulty = stratum_difficulty_for_bits(mine.template.difficulty_bits); 178 self.send_notification("mining.set_difficulty", json!([difficulty])) 179 .await?; 180 self.send_notification( 181 "mining.notify", 182 json!([ 183 job_id, 184 mine.template.prev_hash_hex, 185 mine.template.coinb1_hex(), 186 "", 187 [], 188 mine.template.version_hex, 189 mine.template.nbits_hex, 190 mine.template.ntime_hex, 191 clean_jobs 192 ]), 193 ) 194 .await?; 195 self.jobs.insert(job_id, StratumJob { mine }); 196 Ok(()) 197 } 198 199 async fn handle_submit(&mut self, request: &Value) -> Result<bool> { 200 let params = request 201 .get("params") 202 .and_then(Value::as_array) 203 .context("mining.submit requires params")?; 204 let worker = str_param(params, 0, "worker")?; 205 let job_id = str_param(params, 1, "job id")?; 206 let extranonce2 = hex_array_4(str_param(params, 2, "extranonce2")?)?; 207 let ntime = str_param(params, 3, "ntime")?; 208 let header_nonce = hex_array_4(str_param(params, 4, "nonce")?)?; 209 let authorized = self 210 .authorized_worker 211 .as_deref() 212 .context("worker is not authorized")?; 213 if worker != authorized { 214 bail!("submitted worker does not match authorized worker"); 215 } 216 let job = self 217 .jobs 218 .get(job_id) 219 .cloned() 220 .context("unknown Stratum job")?; 221 if ntime != job.mine.template.ntime_hex { 222 bail!("submitted ntime does not match job"); 223 } 224 225 let (result, outbox) = { 226 let mut node = self.server.node.lock().await; 227 let result = node.submit_external_mine( 228 recipient_from_worker(worker), 229 job.mine.template.clone(), 230 StratumMineShare { 231 extranonce2, 232 header_nonce, 233 }, 234 ); 235 let outbox = node.drain_outbox(); 236 (result, outbox) 237 }; 238 match result { 239 Ok(_) => { 240 self.server.gossip.broadcast(outbox).await?; 241 let worker = worker.to_string(); 242 self.send_job(&worker, false).await?; 243 Ok(true) 244 } 245 Err(error) => Err(error), 246 } 247 } 248 249 async fn send_response(&self, id: Value, result: Value) -> Result<()> { 250 self.send(json!({ "id": id, "result": result, "error": null })) 251 .await 252 } 253 254 async fn send_error(&self, id: Value, code: i64, message: &str) -> Result<()> { 255 self.send(json!({ "id": id, "result": null, "error": [code, message, null] })) 256 .await 257 } 258 259 async fn send_notification(&self, method: &str, params: Value) -> Result<()> { 260 self.send(json!({ "id": null, "method": method, "params": params })) 261 .await 262 } 263 264 async fn send(&self, value: Value) -> Result<()> { 265 let mut writer = self.writer.lock().await; 266 writer 267 .write_all(serde_json::to_string(&value)?.as_bytes()) 268 .await?; 269 writer.write_all(b"\n").await?; 270 Ok(()) 271 } 272 } 273 274 fn str_param<'a>(params: &'a [Value], index: usize, name: &str) -> Result<&'a str> { 275 params 276 .get(index) 277 .and_then(Value::as_str) 278 .with_context(|| format!("mining.submit requires {name}")) 279 } 280 281 fn recipient_from_worker(worker: &str) -> &str { 282 worker 283 .split_once('.') 284 .map_or(worker, |(recipient, _)| recipient) 285 } 286 287 fn hex_array_4(input: &str) -> Result<[u8; 4]> { 288 let bytes = decode_hex(input)?; 289 let len = bytes.len(); 290 bytes 291 .try_into() 292 .map_err(|_| anyhow::anyhow!("expected 4 hex bytes, got {len}")) 293 } 294 295 fn decode_hex(input: &str) -> Result<Vec<u8>> { 296 if input.len() % 2 != 0 { 297 bail!("hex string has odd length"); 298 } 299 let mut bytes = Vec::with_capacity(input.len() / 2); 300 for pair in input.as_bytes().chunks_exact(2) { 301 bytes.push((hex_value(pair[0])? << 4) | hex_value(pair[1])?); 302 } 303 Ok(bytes) 304 } 305 306 fn hex_value(byte: u8) -> Result<u8> { 307 match byte { 308 b'0'..=b'9' => Ok(byte - b'0'), 309 b'a'..=b'f' => Ok(byte - b'a' + 10), 310 b'A'..=b'F' => Ok(byte - b'A' + 10), 311 _ => bail!("invalid hex character"), 312 } 313 } 314 315 fn stratum_difficulty_for_bits(bits: u32) -> f64 { 316 2_f64.powi(bits as i32 - 16).max(0.000001) 317 } 318 319 #[cfg(test)] 320 mod tests { 321 use std::{collections::BTreeMap, net::SocketAddr, sync::Arc}; 322 323 use serde_json::json; 324 use tokio::{ 325 io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, 326 sync::Mutex, 327 }; 328 329 use crate::{ 330 adapters::p2p::GossipNetwork, 331 app::{NodeCore, PeerBook}, 332 domain::{Ledger, StratumMineShare, Wallet}, 333 }; 334 335 use super::StratumServer; 336 337 #[tokio::test] 338 async fn stratum_session_accepts_valid_iuna_share() { 339 let wallet = Wallet::from_seed("stratum-session-wallet"); 340 let ledger = Ledger::new(BTreeMap::new(), 1); 341 let node = Arc::new(Mutex::new(NodeCore::from_ledger(wallet.clone(), ledger, 0))); 342 let peers = Arc::new(Mutex::new(PeerBook::default())); 343 let gossip = GossipNetwork::new_for_tests(Arc::clone(&node), peers); 344 let server = match StratumServer::start( 345 Arc::clone(&node), 346 gossip, 347 "127.0.0.1:0".parse::<SocketAddr>().unwrap(), 348 ) 349 .await 350 { 351 Ok(server) => server, 352 Err(error) if format!("{error:#}").contains("Operation not permitted") => return, 353 Err(error) => panic!("{error:#}"), 354 }; 355 356 let stream = tokio::net::TcpStream::connect(server.listen_addr()) 357 .await 358 .unwrap(); 359 let (read, mut write) = stream.into_split(); 360 let mut lines = BufReader::new(read).lines(); 361 send( 362 &mut write, 363 json!({"id": 1, "method": "mining.subscribe", "params": []}), 364 ) 365 .await; 366 assert_eq!(read_id(&mut lines, 1).await["error"], json!(null)); 367 368 send( 369 &mut write, 370 json!({"id": 2, "method": "mining.authorize", "params": [wallet.address(), "x"]}), 371 ) 372 .await; 373 assert_eq!(read_id(&mut lines, 2).await["result"], json!(true)); 374 let notify = read_method(&mut lines, "mining.notify").await; 375 let job_id = notify["params"][0].as_str().unwrap().to_string(); 376 377 let template = node 378 .lock() 379 .await 380 .external_mine_job(wallet.address(), 1) 381 .unwrap() 382 .template; 383 let mut nonce = None; 384 for candidate in 0_u32..50_000 { 385 let result = node.lock().await.ledger().build_stratum_mine( 386 template.clone(), 387 StratumMineShare { 388 extranonce2: [0, 0, 0, 0], 389 header_nonce: candidate.to_le_bytes(), 390 }, 391 ); 392 if result.is_ok() { 393 nonce = Some(candidate.to_le_bytes()); 394 break; 395 } 396 } 397 let nonce = nonce.expect("expected valid share"); 398 399 send( 400 &mut write, 401 json!({ 402 "id": 3, 403 "method": "mining.submit", 404 "params": [wallet.address(), job_id, "00000000", template.ntime_hex, hex(nonce)] 405 }), 406 ) 407 .await; 408 assert_eq!(read_id(&mut lines, 3).await["result"], json!(true)); 409 let node = node.lock().await; 410 assert_eq!(node.ledger().pending().len(), 1); 411 assert!(node.ledger().pending_blinded_transactions().is_empty()); 412 } 413 414 #[test] 415 fn worker_suffix_is_not_part_of_recipient_address() { 416 assert_eq!(super::recipient_from_worker("abc.bitaxe"), "abc"); 417 assert_eq!(super::recipient_from_worker("abc"), "abc"); 418 } 419 420 async fn send(write: &mut tokio::net::tcp::OwnedWriteHalf, value: serde_json::Value) { 421 write 422 .write_all(serde_json::to_string(&value).unwrap().as_bytes()) 423 .await 424 .unwrap(); 425 write.write_all(b"\n").await.unwrap(); 426 } 427 428 async fn read_id( 429 lines: &mut tokio::io::Lines<BufReader<tokio::net::tcp::OwnedReadHalf>>, 430 id: i64, 431 ) -> serde_json::Value { 432 loop { 433 let line = lines.next_line().await.unwrap().unwrap(); 434 let value: serde_json::Value = serde_json::from_str(&line).unwrap(); 435 if value.get("id").and_then(serde_json::Value::as_i64) == Some(id) { 436 return value; 437 } 438 } 439 } 440 441 async fn read_method( 442 lines: &mut tokio::io::Lines<BufReader<tokio::net::tcp::OwnedReadHalf>>, 443 method: &str, 444 ) -> serde_json::Value { 445 loop { 446 let line = lines.next_line().await.unwrap().unwrap(); 447 let value: serde_json::Value = serde_json::from_str(&line).unwrap(); 448 if value.get("method").and_then(serde_json::Value::as_str) == Some(method) { 449 return value; 450 } 451 } 452 } 453 454 fn hex(bytes: impl AsRef<[u8]>) -> String { 455 const HEX: &[u8; 16] = b"0123456789abcdef"; 456 let mut output = String::new(); 457 for byte in bytes.as_ref() { 458 output.push(HEX[(byte >> 4) as usize] as char); 459 output.push(HEX[(byte & 0x0f) as usize] as char); 460 } 461 output 462 } 463 }