1use crate::config;
2use crate::endpoints::{ElectrumTarget, PublicEndpoint, Wire};
3use crate::error::{Context, Error, ensure};
4use crate::pinning::{Pinned, Pins};
5use crate::transport::{BlockEncoding, Evidence, Provider};
6use bitcoin::{
7 BlockHash, Txid,
8 block::Header,
9 consensus::deserialize,
10 hashes::{Hash, sha256},
11};
12use rustls::{ClientConfig, ClientConnection, ServerName, StreamOwned};
13use serde_json::{Value, json};
14use std::{
15 collections::HashMap,
16 io::{BufRead, BufReader, Read, Write},
17 net::{TcpStream, ToSocketAddrs},
18 sync::{Arc, Mutex},
19 time::{Duration, Instant},
20};
21use urma_chain::{decode as chain_decode, observation::Chain};
22
23enum Stream {
24 Plain(TcpStream),
25 Tls(StreamOwned<ClientConnection, TcpStream>),
26}
27
28impl Stream {
29 fn socket(&self) -> &TcpStream {
30 match self {
31 Self::Plain(socket) => socket,
32 Self::Tls(stream) => &stream.sock,
33 }
34 }
35}
36
37impl Read for Stream {
38 fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
39 match self {
40 Self::Plain(socket) => socket.read(buffer),
41 Self::Tls(stream) => stream.read(buffer),
42 }
43 }
44}
45
46impl Write for Stream {
47 fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
48 match self {
49 Self::Plain(socket) => socket.write(buffer),
50 Self::Tls(stream) => stream.write(buffer),
51 }
52 }
53 fn flush(&mut self) -> std::io::Result<()> {
54 match self {
55 Self::Plain(socket) => socket.flush(),
56 Self::Tls(stream) => stream.flush(),
57 }
58 }
59}
60
61struct Session {
62 stream: BufReader<Stream>,
63 next_id: u64,
64 unspent: HashMap<String, (Instant, Vec<Value>)>,
65}
66
67impl Session {
68 fn open(target: &ElectrumTarget, pins: &Arc<Pins>) -> Result<Self, Error> {
69 let socket = connect(target)?;
70 let stream = match target.wire {
71 Wire::PlainTcp => {
72 tracing::warn!(host = %target.host, port = target.port, "electrum endpoint explicitly configured without TLS");
73 Stream::Plain(socket)
74 }
75 Wire::Tls => {
76 let verifier = Pinned {
77 pins: pins.clone(),
78 key: format!("{}_{}", target.host, target.port),
79 };
80 let tls = ClientConfig::builder()
81 .with_safe_defaults()
82 .with_custom_certificate_verifier(Arc::new(verifier))
83 .with_no_client_auth();
84 let name = ServerName::try_from(target.host.as_str())
85 .map_err(|cause| Error::Invalid(format!("electrum host name: {cause}")))?;
86 let connection = ClientConnection::new(Arc::new(tls), name)?;
87 Stream::Tls(StreamOwned::new(connection, socket))
88 }
89 };
90 Ok(Self {
91 stream: BufReader::new(stream),
92 next_id: 1,
93 unspent: HashMap::new(),
94 })
95 }
96
97 fn request(&mut self, method: &str, params: Value, timeout: Duration) -> Result<Value, Error> {
98 let id = self.next_id;
99 self.next_id = id
100 .checked_add(1)
101 .context("electrum request counter overflow")?;
102 let socket = self.stream.get_ref().socket();
103 socket.set_read_timeout(Some(timeout))?;
104 socket.set_write_timeout(Some(timeout))?;
105 let mut line =
106 serde_json::to_vec(&json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}))?;
107 line.push(b'\n');
108 let stream = self.stream.get_mut();
109 stream.write_all(&line)?;
110 stream.flush()?;
111 loop {
112 let reply = self.read_line()?;
113 if reply["id"] != json!(id) {
114 tracing::debug!(method = %reply["method"], "skipping electrum notification");
115 continue;
116 }
117 if !reply["error"].is_null() {
118 return Err(server_error(method, &reply["error"]));
119 }
120 return Ok(reply
121 .get("result")
122 .context("electrum reply omitted result")?
123 .clone());
124 }
125 }
126
127 fn read_line(&mut self) -> Result<Value, Error> {
128 let mut buffer = Vec::new();
129 let limit = config::ELECTRUM_MAX_LINE_BYTES;
130 let bound = u64::try_from(limit)?
131 .checked_add(1)
132 .context("line bound overflow")?;
133 let count = (&mut self.stream)
134 .take(bound)
135 .read_until(b'\n', &mut buffer)?;
136 if count == 0 {
137 return Err(Error::Io(std::io::Error::new(
138 std::io::ErrorKind::UnexpectedEof,
139 "electrum server closed the connection",
140 )));
141 }
142 ensure!(buffer.len() <= limit, "electrum reply exceeds capacity");
143 ensure!(
144 buffer.last() == Some(&b'\n'),
145 "electrum reply was truncated"
146 );
147 Ok(serde_json::from_slice(&buffer)?)
148 }
149}
150
151fn connect(target: &ElectrumTarget) -> Result<TcpStream, Error> {
152 let mut last = Error::Missing(format!("{} resolved to no address", target.host));
153 for address in (target.host.as_str(), target.port).to_socket_addrs()? {
154 match TcpStream::connect_timeout(&address, config::ELECTRUM_CONNECT_TIMEOUT) {
155 Ok(socket) => {
156 socket.set_nodelay(true)?;
157 return Ok(socket);
158 }
159 Err(error) => {
160 tracing::warn!(%address, %error, "electrum address unreachable");
161 last = Error::Io(error);
162 }
163 }
164 }
165 Err(last)
166}
167
168fn server_error(method: &str, error: &Value) -> Error {
169 let message = error["message"].to_string();
170 let lowered = message.to_ascii_lowercase();
171 if lowered.contains("no such mempool or blockchain transaction")
172 || lowered.contains("not found")
173 {
174 return Error::Missing(config::ABSENT_TRANSACTION.into());
175 }
176 Error::Unsupported(format!("electrum {method} failed: {message}"))
177}
178
179enum Link {
180 Closed,
181 Open(Session),
182}
183
184pub struct Electrum {
185 endpoint: PublicEndpoint,
186 target: ElectrumTarget,
187 pins: Arc<Pins>,
188 link: Mutex<Link>,
189 seen: Mutex<HashMap<BlockHash, u64>>,
190}
191
192impl Electrum {
193 pub fn new(endpoint: PublicEndpoint, pins: Arc<Pins>) -> Result<Self, Error> {
194 let target = endpoint.electrum_target()?;
195 Ok(Self {
196 endpoint,
197 target,
198 pins,
199 link: Mutex::new(Link::Closed),
200 seen: Mutex::new(HashMap::new()),
201 })
202 }
203
204 fn session<T>(
205 &self,
206 chain: Chain,
207 work: impl Fn(&mut Session) -> Result<T, Error>,
208 ) -> Result<T, Error> {
209 let mut link = match self.link.lock() {
210 Ok(link) => link,
211 Err(poisoned) => {
212 tracing::error!(endpoint = %self.endpoint.url(), "electrum link lock poisoned");
213 poisoned.into_inner()
214 }
215 };
216 let mut fresh = false;
217 loop {
218 if matches!(*link, Link::Closed) {
219 let mut session = Session::open(&self.target, &self.pins)?;
220 match self.handshake(&mut session, chain) {
221 Ok(()) => *link = Link::Open(session),
222 Err(error) => {
223 tracing::warn!(endpoint = %self.endpoint.url(), %error, "electrum handshake failed");
224 return Err(error);
225 }
226 }
227 fresh = true;
228 }
229 let Link::Open(session) = &mut *link else {
230 return Err(Error::Invalid("electrum link closed during request".into()));
231 };
232 match work(session) {
233 Ok(value) => return Ok(value),
234 Err(Error::Io(cause)) if fresh => {
235 tracing::warn!(endpoint = %self.endpoint.url(), error = %cause, "electrum connection dropped on a fresh session");
236 *link = Link::Closed;
237 return Err(Error::Io(cause));
238 }
239 Err(Error::Io(cause)) => {
240 tracing::warn!(endpoint = %self.endpoint.url(), error = %cause, "electrum connection dropped; reconnecting and retrying once");
241 *link = Link::Closed;
242 }
243 Err(error) => return Err(error),
244 }
245 }
246 }
247
248 fn handshake(&self, session: &mut Session, chain: Chain) -> Result<(), Error> {
249 let timeout = config::method_timeout("getblockchaininfo");
250 let version = session.request(
251 "server.version",
252 json!(["urma/0.2", config::ELECTRUM_PROTOCOL]),
253 timeout,
254 )?;
255 ensure!(
256 version[1] == json!(config::ELECTRUM_PROTOCOL),
257 "electrum server negotiated protocol {} instead of {}",
258 version[1],
259 config::ELECTRUM_PROTOCOL
260 );
261 let genesis =
262 header_hash(&session.request("blockchain.block.header", json!([0]), timeout)?)?;
263 ensure!(
264 genesis == chain.genesis()?.0,
265 "electrum server genesis does not match {}",
266 chain.label()
267 );
268 tracing::debug!(endpoint = %self.endpoint.url(), server = %version[0], "electrum session verified");
269 Ok(())
270 }
271
272 fn remember(&self, hash: BlockHash, height: u64) {
273 match self.seen.lock() {
274 Ok(mut seen) => {
275 seen.insert(hash, height);
276 }
277 Err(poisoned) => {
278 tracing::error!(endpoint = %self.endpoint.url(), "electrum header table lock poisoned");
279 poisoned.into_inner().insert(hash, height);
280 }
281 }
282 }
283
284 fn seen_height(&self, hash: BlockHash) -> Result<u64, Error> {
285 let seen = match self.seen.lock() {
286 Ok(seen) => seen,
287 Err(poisoned) => {
288 tracing::error!(endpoint = %self.endpoint.url(), "electrum header table lock poisoned");
289 poisoned.into_inner()
290 }
291 };
292 match seen.get(&hash) {
293 Some(height) => Ok(*height),
294 None => Err(Error::Unsupported(format!(
295 "electrum server has not served block {hash} to this session"
296 ))),
297 }
298 }
299
300 fn tip(&self, session: &mut Session) -> Result<(u64, BlockHash), Error> {
301 let timeout = config::method_timeout("getblockchaininfo");
302 let tip = session.request("blockchain.headers.subscribe", json!([]), timeout)?;
303 let height = tip["height"]
304 .as_u64()
305 .context("electrum tip height missing")?;
306 let hash = header_hash(&tip["hex"])?;
307 self.remember(hash, height);
308 Ok((height, hash))
309 }
310
311 fn block_hash(&self, session: &mut Session, height: u64) -> Result<BlockHash, Error> {
312 let timeout = config::method_timeout("getblockhash");
313 let header = session.request("blockchain.block.header", json!([height]), timeout)?;
314 let hash = header_hash(&header)?;
315 self.remember(hash, height);
316 Ok(hash)
317 }
318
319 fn raw_transaction(&self, session: &mut Session, txid: &Txid) -> Result<String, Error> {
320 let timeout = config::method_timeout("getrawtransaction");
321 let raw = session.request("blockchain.transaction.get", json!([txid]), timeout)?;
322 Ok(raw
323 .as_str()
324 .context("electrum transaction bytes missing")?
325 .to_owned())
326 }
327
328 fn verbose_transaction(&self, session: &mut Session, txid: &Txid) -> Result<Value, Error> {
329 let timeout = config::method_timeout("getrawtransaction");
330 let verbose =
331 session.request("blockchain.transaction.get", json!([txid, true]), timeout)?;
332 let confirmations = config::confirmations(&verbose)?;
333 if confirmations <= 0 {
334 return Ok(json!({"confirmations":0}));
335 }
336 let hash: BlockHash = verbose["blockhash"]
337 .as_str()
338 .context("electrum inclusion block missing")?
339 .parse()?;
340 let (tip, _) = self.tip(session)?;
341 let depth = u64::try_from(confirmations)?;
342 let height = tip
343 .checked_add(1)
344 .and_then(|next| next.checked_sub(depth))
345 .context("electrum confirmations exceed tip height")?;
346 ensure!(
347 self.block_hash(session, height)? == hash,
348 "electrum inclusion block disagrees with its header chain"
349 );
350 Ok(json!({"confirmations":confirmations,"blockhash":hash}))
351 }
352
353 fn listunspent(
354 &self,
355 session: &mut Session,
356 script: &bitcoin::Script,
357 ) -> Result<Vec<Value>, Error> {
358 let key = scripthash(script);
359 let now = Instant::now();
360 let cached = session.unspent.get(&key);
361 for (fetched, rows) in cached.iter() {
362 if now.duration_since(*fetched) < config::ELECTRUM_UNSPENT_CACHE {
363 tracing::debug!(endpoint = %self.endpoint.url(), rows = rows.len(), "unspent list served from the session cache");
364 return Ok(rows.clone());
365 }
366 }
367 let timeout = config::method_timeout("addressutxos");
368 let rows = session.request("blockchain.scripthash.listunspent", json!([key]), timeout)?;
369 let rows = rows.as_array().context("electrum unspent list missing")?;
370 ensure!(
371 rows.len() <= config::SOURCE_MAX_UTXOS,
372 "UTXO client capacity exceeded"
373 );
374 session.unspent.insert(key, (now, rows.clone()));
375 Ok(rows.clone())
376 }
377
378 fn address_utxos(
379 &self,
380 session: &mut Session,
381 chain: Chain,
382 address: &str,
383 ) -> Result<Value, Error> {
384 let script = urma_wallet::address::destination_script(address, chain)?;
385 let mut outputs = Vec::new();
386 for row in self.listunspent(session, &script)? {
387 let height = row["height"]
388 .as_u64()
389 .context("electrum unspent height missing")?;
390 outputs.push(json!({
391 "txid": row["tx_hash"],
392 "vout": row["tx_pos"],
393 "value": row["value"],
394 "status": {"confirmed": height > 0, "block_height": height},
395 }));
396 }
397 Ok(Value::Array(outputs))
398 }
399
400 fn txout(&self, session: &mut Session, chain: Chain, args: &[Value]) -> Result<Value, Error> {
401 let txid: Txid = args
402 .first()
403 .context("missing txid")?
404 .as_str()
405 .context("invalid txid")?
406 .parse()?;
407 let vout = args
408 .get(1)
409 .context("missing vout")?
410 .as_u64()
411 .context("invalid vout")?;
412 let raw = self.raw_transaction(session, &txid)?;
413 let transaction = chain_decode::transaction(&hex::decode(raw)?, chain)?;
414 ensure!(
415 transaction.compute_txid() == txid,
416 "electrum transaction hash mismatch"
417 );
418 let output = transaction
419 .output
420 .get(usize::try_from(vout)?)
421 .context("transaction output index out of range")?;
422 let mut unspent = None;
423 for row in self.listunspent(session, &output.script_pubkey)? {
424 if row["tx_hash"] == json!(txid) && row["tx_pos"] == json!(vout) {
425 unspent = Some(row);
426 }
427 }
428 let Some(row) = unspent else {
429 return Ok(Value::Null);
430 };
431 let height = row["height"]
432 .as_u64()
433 .context("electrum unspent height missing")?;
434 let confirmations = if height > 0 {
435 let (tip, _) = self.tip(session)?;
436 tip.checked_sub(height)
437 .and_then(|n| n.checked_add(1))
438 .context("chain changed during output lookup")?
439 } else {
440 0
441 };
442 Ok(json!({
443 "confirmations": confirmations,
444 "coinbase": transaction.is_coinbase(),
445 "value": serde_json::from_str::<Value>(&output.value.to_string_in(bitcoin::Denomination::Bitcoin))?,
446 "scriptPubKey": {"hex": hex::encode(output.script_pubkey.as_bytes())},
447 }))
448 }
449
450 fn dispatch(
451 &self,
452 session: &mut Session,
453 chain: Chain,
454 method: &str,
455 args: &[Value],
456 ) -> Result<Value, Error> {
457 match method {
458 "getblockchaininfo" => {
459 let (height, hash) = self.tip(session)?;
460 Ok(json!({"blocks":height,"bestblockhash":hash,"initialblockdownload":false}))
461 }
462 "getblockhash" => {
463 let height = args
464 .first()
465 .context("missing block height")?
466 .as_u64()
467 .context("invalid block height")?;
468 Ok(json!(self.block_hash(session, height)?))
469 }
470 "getblockheader" => {
471 let hash: BlockHash = args
472 .first()
473 .context("missing block hash")?
474 .as_str()
475 .context("invalid block hash")?
476 .parse()?;
477 let height = self.seen_height(hash)?;
478 Ok(json!({"hash":hash,"height":height}))
479 }
480 "getrawtransaction" => {
481 let txid: Txid = args
482 .first()
483 .context("missing txid")?
484 .as_str()
485 .context("invalid txid")?
486 .parse()?;
487 if args.get(1).context("missing transaction verbosity")? == &json!(true) {
488 return self.verbose_transaction(session, &txid);
489 }
490 Ok(json!(self.raw_transaction(session, &txid)?))
491 }
492 "gettxout" => self.txout(session, chain, args),
493 "addressutxos" => {
494 let address = args
495 .first()
496 .context("missing address")?
497 .as_str()
498 .context("invalid address")?;
499 self.address_utxos(session, chain, address)
500 }
501 "sendrawtransaction" => {
502 let raw = args.first().context("missing broadcast transaction")?;
503 let timeout = config::method_timeout(method);
504 session.unspent.clear();
505 session.request("blockchain.transaction.broadcast", json!([raw]), timeout)
506 }
507 other => Err(Error::Unsupported(format!(
508 "electrum servers do not serve {other}"
509 ))),
510 }
511 }
512}
513
514impl Electrum {
515 pub fn probe(&self, chain: Chain, method: &str, params: Value) -> Result<Value, Error> {
516 ensure!(
517 config::ELECTRUM_PROBE_METHODS.contains(&method),
518 "{method} is not a read-only electrum probe"
519 );
520 let timeout = config::method_timeout("addressutxos");
521 self.session(chain, |session| {
522 session.request(method, params.clone(), timeout)
523 })
524 }
525}
526
527impl Provider for Electrum {
528 fn label(&self) -> String {
529 self.endpoint.url().to_owned()
530 }
531
532 fn evidence(&self) -> Evidence {
533 Evidence::PublicProviderObservation
534 }
535
536 fn block_encoding(&self) -> BlockEncoding {
537 BlockEncoding::Core
538 }
539
540 fn supports(&self, method: &str) -> bool {
541 !matches!(method, "getblock" | "getrawmempool" | "testmempoolaccept")
542 }
543
544 fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
545 self.session(chain, |session| self.dispatch(session, chain, method, args))
546 }
547}
548
549fn header_hash(value: &Value) -> Result<BlockHash, Error> {
550 let bytes = hex::decode(value.as_str().context("electrum header bytes missing")?)?;
551 ensure!(bytes.len() == 80, "electrum header is not 80 bytes");
552 let header: Header = deserialize(&bytes)?;
553 Ok(header.block_hash())
554}
555
556fn scripthash(script: &bitcoin::Script) -> String {
557 let mut digest = sha256::Hash::hash(script.as_bytes()).to_byte_array();
558 digest.reverse();
559 hex::encode(digest)
560}