netget/server/stratum/
wire.rs

1//! Stratum V1 on the wire, for both roles: newline-delimited JSON-RPC messages, and the share
2//! arithmetic a pool and a miner must agree on byte for byte — the coinbase a job describes,
3//! the block header a share commits to, its double SHA-256, and its difficulty.
4use anyhow::{bail, ensure, Context, Result};
5use serde_json::{json, Value};
6use sha2::{Digest, Sha256};
7use std::time::Duration;
8use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader};
9
10/// One JSON-RPC line, at most.
11pub const MAX_LINE: usize = 16 * 1024;
12pub const IO_TIMEOUT: Duration = Duration::from_secs(30);
13/// Bytes of extranonce2 the pool asks miners to roll.
14pub const EXTRANONCE2_SIZE: usize = 4;
15/// Bytes of extranonce1 the pool assigns each connection.
16pub const EXTRANONCE1_SIZE: usize = 4;
17/// How far past the job's ntime a share may roll its timestamp, in seconds.
18pub const NTIME_ROLL: u32 = 7200;
19/// Merkle branches a job may carry (a block of 2^16 transactions).
20pub const MAX_BRANCHES: usize = 16;
21/// Coinbase halves are at most this long.
22pub const MAX_COINBASE: usize = 4096;
23/// The coinbase message a pool job carries, in bytes.
24pub const MAX_COINBASE_MESSAGE: usize = 64;
25/// Difficulty 1: the target 0x00000000FFFF0000…, as a number.
26pub const DIFF1: f64 = 26959535291011309493156476344723991336010898738574164086137773096960.0;
27
28/// Newline-delimited JSON values from one side of a connection.
29pub struct Lines<R> {
30    reader: BufReader<R>,
31    line: Vec<u8>,
32}
33
34impl<R: AsyncRead + Unpin> Lines<R> {
35    pub fn new(reader: R) -> Self {
36        Self {
37            reader: BufReader::new(reader),
38            line: Vec::new(),
39        }
40    }
41
42    /// The next JSON value; None at a clean end of stream. A line over `MAX_LINE`, or one
43    /// that is not JSON, is an error: the peer is not speaking Stratum.
44    pub async fn next(&mut self, idle: Duration) -> Result<Option<Value>> {
45        loop {
46            self.line.clear();
47            loop {
48                let buf = tokio::time::timeout(idle, self.reader.fill_buf())
49                    .await
50                    .context("Stratum peer idle")??;
51                if buf.is_empty() {
52                    if self.line.is_empty() {
53                        return Ok(None);
54                    }
55                    bail!("Stratum stream ended inside a line");
56                }
57                let (take, done) = match buf.iter().position(|b| *b == b'\n') {
58                    Some(i) => (i + 1, true),
59                    None => (buf.len(), false),
60                };
61                let content = self.line.len() + take - usize::from(done);
62                ensure!(
63                    content <= MAX_LINE,
64                    "Stratum line longer than {MAX_LINE} bytes"
65                );
66                self.line.extend_from_slice(&buf[..take]);
67                self.reader.consume(take);
68                if done {
69                    break;
70                }
71            }
72            let text = std::str::from_utf8(&self.line).context("Stratum line is not UTF-8")?;
73            if text.trim().is_empty() {
74                continue;
75            }
76            return serde_json::from_str(text.trim())
77                .map(Some)
78                .context("Stratum line is not JSON");
79        }
80    }
81}
82
83pub enum Message {
84    Request {
85        id: Value,
86        method: String,
87        params: Value,
88    },
89    Notification {
90        method: String,
91        params: Value,
92    },
93    Response {
94        id: Value,
95        result: Value,
96        error: Value,
97    },
98}
99
100pub fn parse(v: Value) -> Result<Message> {
101    let obj = v
102        .as_object()
103        .context("a Stratum message is a JSON object")?;
104    let id = obj.get("id").cloned().unwrap_or(Value::Null);
105    match obj.get("method") {
106        Some(Value::String(method)) => {
107            let params = obj.get("params").cloned().unwrap_or(json!([]));
108            Ok(if id.is_null() {
109                Message::Notification {
110                    method: method.clone(),
111                    params,
112                }
113            } else {
114                Message::Request {
115                    id,
116                    method: method.clone(),
117                    params,
118                }
119            })
120        }
121        Some(_) => bail!("method must be a string"),
122        None => Ok(Message::Response {
123            id,
124            result: obj.get("result").cloned().unwrap_or(Value::Null),
125            error: obj.get("error").cloned().unwrap_or(Value::Null),
126        }),
127    }
128}
129
130pub fn line(v: &Value) -> Vec<u8> {
131    let mut out = serde_json::to_vec(v).unwrap_or_default();
132    out.push(b'\n');
133    out
134}
135
136/// `[code, message, null]`, Stratum's error shape (20 other, 21 job not found, 22 duplicate,
137/// 23 low difficulty, 24 unauthorized worker, 25 not subscribed).
138pub fn error(code: u32, message: &str) -> Value {
139    json!([code, message, null])
140}
141
142pub fn sha256d(data: &[u8]) -> [u8; 32] {
143    Sha256::digest(Sha256::digest(data)).into()
144}
145
146/// A hash as block explorers and RPC show it: byte-reversed hex.
147pub fn display_hex(hash: &[u8; 32]) -> String {
148    let mut h = *hash;
149    h.reverse();
150    hex::encode(h)
151}
152
153/// The difficulty a header hash meets: difficulty 1's target over the hash as a number.
154pub fn difficulty(hash: &[u8; 32]) -> f64 {
155    let value = hash
156        .iter()
157        .rev()
158        .fold(0f64, |acc, b| acc * 256.0 + f64::from(*b));
159    if value == 0.0 {
160        f64::INFINITY
161    } else {
162        DIFF1 / value
163    }
164}
165
166/// A 32-bit field as Stratum writes it: big-endian hex of the number.
167pub fn u32_hex(v: u32) -> String {
168    format!("{v:08x}")
169}
170
171pub fn parse_u32_hex(s: &str, what: &str) -> Result<u32> {
172    ensure!(s.len() == 8, "{what} must be 8 hex digits");
173    u32::from_str_radix(s, 16).with_context(|| format!("{what} is not hex"))
174}
175
176/// One unit of work: what `mining.notify` carries.
177#[derive(Clone, Debug)]
178pub struct Job {
179    pub job_id: String,
180    /// The previous block hash in header byte order.
181    pub prevhash: [u8; 32],
182    pub coinb1: Vec<u8>,
183    pub coinb2: Vec<u8>,
184    pub branches: Vec<[u8; 32]>,
185    pub version: u32,
186    pub nbits: u32,
187    pub ntime: u32,
188    pub clean: bool,
189}
190
191/// Stratum's prevhash: header order with every 32-bit word byte-swapped.
192fn swap_words(b: &[u8; 32]) -> [u8; 32] {
193    let mut out = [0u8; 32];
194    for i in 0..8 {
195        for j in 0..4 {
196            out[i * 4 + j] = b[i * 4 + 3 - j];
197        }
198    }
199    out
200}
201
202fn hex32(s: &str, what: &str) -> Result<[u8; 32]> {
203    let v = hex::decode(s).with_context(|| format!("{what} is not hex"))?;
204    v.try_into()
205        .map_err(|_| anyhow::anyhow!("{what} must be 32 bytes"))
206}
207
208impl Job {
209    pub fn from_notify(p: &Value) -> Result<Job> {
210        let a = p
211            .as_array()
212            .context("mining.notify params must be an array")?;
213        ensure!(a.len() >= 9, "mining.notify has {} params, not 9", a.len());
214        let s = |i: usize, what: &str| -> Result<&str> {
215            a[i].as_str()
216                .with_context(|| format!("{what} must be a string"))
217        };
218        let coinb1 = hex::decode(s(2, "coinb1")?).context("coinb1 is not hex")?;
219        let coinb2 = hex::decode(s(3, "coinb2")?).context("coinb2 is not hex")?;
220        ensure!(
221            coinb1.len() <= MAX_COINBASE && coinb2.len() <= MAX_COINBASE,
222            "coinbase longer than {MAX_COINBASE} bytes"
223        );
224        let branches = a[4]
225            .as_array()
226            .context("merkle branches must be an array")?
227            .iter()
228            .map(|b| hex32(b.as_str().unwrap_or_default(), "merkle branch"))
229            .collect::<Result<Vec<_>>>()?;
230        ensure!(
231            branches.len() <= MAX_BRANCHES,
232            "more than {MAX_BRANCHES} merkle branches"
233        );
234        Ok(Job {
235            job_id: s(0, "job_id")?.to_string(),
236            prevhash: swap_words(&hex32(s(1, "prevhash")?, "prevhash")?),
237            coinb1,
238            coinb2,
239            branches,
240            version: parse_u32_hex(s(5, "version")?, "version")?,
241            nbits: parse_u32_hex(s(6, "nbits")?, "nbits")?,
242            ntime: parse_u32_hex(s(7, "ntime")?, "ntime")?,
243            clean: a[8].as_bool().unwrap_or(false),
244        })
245    }
246
247    pub fn notify_params(&self) -> Value {
248        json!([
249            self.job_id,
250            hex::encode(swap_words(&self.prevhash)),
251            hex::encode(&self.coinb1),
252            hex::encode(&self.coinb2),
253            self.branches.iter().map(hex::encode).collect::<Vec<_>>(),
254            u32_hex(self.version),
255            u32_hex(self.nbits),
256            u32_hex(self.ntime),
257            self.clean,
258        ])
259    }
260
261    /// The previous block hash as RPC shows it.
262    pub fn prev_display(&self) -> String {
263        display_hex(&self.prevhash)
264    }
265
266    pub fn merkle_root(&self, extranonce1: &[u8], extranonce2: &[u8]) -> [u8; 32] {
267        let mut coinbase = Vec::with_capacity(
268            self.coinb1.len() + extranonce1.len() + extranonce2.len() + self.coinb2.len(),
269        );
270        coinbase.extend_from_slice(&self.coinb1);
271        coinbase.extend_from_slice(extranonce1);
272        coinbase.extend_from_slice(extranonce2);
273        coinbase.extend_from_slice(&self.coinb2);
274        let mut root = sha256d(&coinbase);
275        for b in &self.branches {
276            let mut pair = [0u8; 64];
277            pair[..32].copy_from_slice(&root);
278            pair[32..].copy_from_slice(b);
279            root = sha256d(&pair);
280        }
281        root
282    }
283
284    /// The 80-byte header for a share, with the nonce left zero.
285    pub fn header(&self, merkle_root: &[u8; 32], ntime: u32) -> [u8; 80] {
286        let mut h = [0u8; 80];
287        h[0..4].copy_from_slice(&self.version.to_le_bytes());
288        h[4..36].copy_from_slice(&self.prevhash);
289        h[36..68].copy_from_slice(merkle_root);
290        h[68..72].copy_from_slice(&ntime.to_le_bytes());
291        h[72..76].copy_from_slice(&self.nbits.to_le_bytes());
292        h
293    }
294
295    /// The hash a share (extranonce2, ntime, nonce) commits to.
296    pub fn share_hash(
297        &self,
298        extranonce1: &[u8],
299        extranonce2: &[u8],
300        ntime: u32,
301        nonce: u32,
302    ) -> [u8; 32] {
303        let mut h = self.header(&self.merkle_root(extranonce1, extranonce2), ntime);
304        h[76..80].copy_from_slice(&nonce.to_le_bytes());
305        sha256d(&h)
306    }
307}
308
309/// A script push of `data`.
310fn push(script: &mut Vec<u8>, data: &[u8]) {
311    script.push(data.len() as u8);
312    script.extend_from_slice(data);
313}
314
315/// A pool's job: a coinbase carrying the block height (BIP 34) and `message`, paying nothing
316/// to an OP_RETURN output, with room for the extranonces; no other transactions.
317pub fn pool_job(
318    job_id: String,
319    prev_display: &[u8; 32],
320    height: u32,
321    message: &str,
322    nbits: u32,
323    ntime: u32,
324    clean: bool,
325) -> Job {
326    let mut prevhash = *prev_display;
327    prevhash.reverse();
328    let mut height_bytes = height.to_le_bytes().to_vec();
329    while height_bytes.len() > 1 && height_bytes.last() == Some(&0) {
330        height_bytes.pop();
331    }
332    if height_bytes.last().is_some_and(|b| b & 0x80 != 0) {
333        height_bytes.push(0);
334    }
335    let message = &message.as_bytes()[..message.len().min(MAX_COINBASE_MESSAGE)];
336    let mut script = Vec::new();
337    push(&mut script, &height_bytes);
338    push(&mut script, message);
339    let script_len = script.len() + 1 + EXTRANONCE1_SIZE + EXTRANONCE2_SIZE;
340    let mut coinb1 = vec![1, 0, 0, 0, 1];
341    coinb1.extend_from_slice(&[0u8; 32]);
342    coinb1.extend_from_slice(&[0xff; 4]);
343    coinb1.push(script_len as u8);
344    coinb1.extend_from_slice(&script);
345    coinb1.push((EXTRANONCE1_SIZE + EXTRANONCE2_SIZE) as u8);
346    let mut coinb2 = vec![0xff; 4];
347    coinb2.push(1);
348    coinb2.extend_from_slice(&0u64.to_le_bytes());
349    let out_script = [&[0x6a, 6][..], b"netget"].concat();
350    coinb2.push(out_script.len() as u8);
351    coinb2.extend_from_slice(&out_script);
352    coinb2.extend_from_slice(&[0u8; 4]);
353    Job {
354        job_id,
355        prevhash,
356        coinb1,
357        coinb2,
358        branches: Vec::new(),
359        version: 0x2000_0000,
360        nbits,
361        ntime,
362        clean,
363    }
364}
365
366/// A previous block hash as RPC writes it (64 hex digits), or zero when absent.
367pub fn parse_prev_display(s: Option<&str>) -> Result<[u8; 32]> {
368    match s {
369        None | Some("") => Ok([0u8; 32]),
370        Some(s) => hex32(s, "prev_hash"),
371    }
372}
netget/server/stratum/wire.rs

netget/server/stratum/
wire.rs
