1use crate::config::Limits;
2use crate::config::{
3 SOURCE_MAX_BLOCK_BYTES, SOURCE_MAX_JSON_BYTES, SOURCE_MAX_SCAN_BLOCKS, SOURCE_MAX_SCAN_BYTES,
4 SOURCE_MAX_UTXOS,
5};
6use crate::error::{Context, Error, bail, ensure};
7use crate::http;
8use crate::journal::{self, Plan};
9use crate::{
10 backend::{self, Evidence, Family, Locator, Observation, RecordSource},
11 bitcoin_rpc::{self, Scan},
12 container,
13};
14use bitcoin::{
15 Address, Block, BlockHash, Network, Transaction, Txid, consensus::deserialize, hashes::Hash,
16};
17use serde::Deserialize;
18use serde_json::{Value, json};
19use std::str::FromStr;
20use urma_core::envelope;
21use urma_wallet::funding::Funding;
22
23pub struct Source {
24 url: String,
25 network: Network,
26 genesis: BlockHash,
27}
28
29#[derive(Clone, Deserialize)]
30struct TxStatus {
31 confirmed: bool,
32 block_height: Option<u64>,
33 block_hash: Option<BlockHash>,
34}
35
36#[derive(Deserialize)]
37struct Utxo {
38 txid: Txid,
39 vout: u32,
40 value: u64,
41 status: TxStatus,
42}
43
44impl Source {
45 pub(crate) fn endpoint(&self) -> &str {
46 &self.url
47 }
48
49 pub(crate) fn confirmed_funding(&self, plan: &Plan) -> Result<(), Error> {
50 ensure!(
51 plan.network == self.network.to_string(),
52 "funding network mismatch"
53 );
54 let raw = hex::decode(&plan.funding.raw_transaction)?;
55 let expected: Transaction = deserialize(&raw)?;
56 let txid = expected.compute_txid();
57 let tip = self.height()?;
58 let tip_hash = self.block_hash(tip)?;
59 let (found, found_raw) = self.transaction(txid)?;
60 ensure!(
61 found == expected && found_raw == raw,
62 "funding transaction/witness differs from journal"
63 );
64 ensure!(
65 usize::try_from(plan.funding.vout)? < found.output.len(),
66 "funding vout absent"
67 );
68 let status: TxStatus = self.json(&format!("/tx/{txid}/status"))?;
69 ensure!(
70 self.verify_confirmation(&found, &status, tip)? >= 1,
71 "funding must be confirmed before relay"
72 );
73 self.require_unspent(txid, plan.funding.vout)?;
74 ensure!(
75 self.block_hash(tip)? == tip_hash,
76 "source chain changed during relay preflight"
77 );
78 Ok(())
79 }
80
81 pub(crate) fn require_unspent(&self, txid: Txid, vout: u32) -> Result<(), Error> {
82 let status: Value = self.json(&format!("/tx/{txid}/outspend/{vout}"))?;
83 ensure!(
84 status["spent"].as_bool() == Some(false),
85 "relay input is spent or unspentness is unknown"
86 );
87 Ok(())
88 }
89
90 pub fn connect(url: &str, network: &Network) -> Result<Self, Error> {
91 ensure!(
92 matches!(network, Network::Regtest | Network::Testnet4),
93 "only regtest and testnet4 are supported"
94 );
95 let url = checked_url(url)?;
96 let genesis = bitcoin::blockdata::constants::genesis_block(*network).block_hash();
97 let source = Self {
98 url,
99 network: *network,
100 genesis,
101 };
102 ensure!(
103 source.block_hash(0)? == genesis,
104 "source genesis differs from requested network"
105 );
106 Ok(source)
107 }
108
109 fn get_optional(&self, path: &str, limit: usize) -> Result<Option<Vec<u8>>, Error> {
110 let response = minreq::get(format!("{}{path}", self.url))
111 .with_timeout(20)
112 .with_max_redirects(0)
113 .with_max_headers_size(16_384)
114 .with_max_status_line_length(1_024)
115 .with_header("Cache-Control", "no-cache")
116 .send_lazy()
117 .with_context(|| format!("source GET {path}"))?;
118 if response.status_code == 404 {
119 return Ok(None);
120 }
121 ensure!(
122 response.status_code == 200,
123 "source GET {path}: HTTP {}",
124 response.status_code
125 );
126 let length_header = response.headers.get("content-length");
127 for length in length_header.iter() {
128 ensure!(
129 length
130 .parse::<usize>()
131 .context("invalid source content length")?
132 <= limit,
133 "source response exceeds byte limit"
134 );
135 }
136 let bytes = http::body(response, limit)?;
137 ensure!(bytes.len() <= limit, "source response exceeds byte limit");
138 Ok(Some(bytes))
139 }
140
141 fn get(&self, path: &str, limit: usize) -> Result<Vec<u8>, Error> {
142 self.get_optional(path, limit)?
143 .with_context(|| format!("source GET {path}: not found"))
144 }
145
146 fn json<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T, Error> {
147 serde_json::from_slice(&self.get(path, SOURCE_MAX_JSON_BYTES)?)
148 .with_context(|| format!("invalid source JSON: {path}"))
149 }
150
151 pub fn height(&self) -> Result<u64, Error> {
152 std::str::from_utf8(&self.get("/blocks/tip/height", 32)?)?
153 .trim()
154 .parse()
155 .context("invalid source tip height")
156 }
157
158 pub fn block_hash(&self, height: u64) -> Result<BlockHash, Error> {
159 BlockHash::from_str(
160 std::str::from_utf8(&self.get(&format!("/block-height/{height}"), 128)?)?.trim(),
161 )
162 .context("invalid source block hash")
163 }
164
165 fn block(&self, hash: BlockHash) -> Result<(Block, usize), Error> {
166 let bytes = self.get(&format!("/block/{hash}/raw"), SOURCE_MAX_BLOCK_BYTES)?;
167 let block =
168 urma_chain::decode::bitcoin_block(&bytes, hash).context("invalid source raw block")?;
169 bitcoin_rpc::validate_block_for_network(&block, hash, self.network)?;
170 Ok((block, bytes.len()))
171 }
172
173 fn transaction(&self, txid: Txid) -> Result<(Transaction, Vec<u8>), Error> {
174 self.transaction_optional(txid)?
175 .with_context(|| format!("source transaction {txid}: not found"))
176 }
177
178 fn transaction_optional(&self, txid: Txid) -> Result<Option<(Transaction, Vec<u8>)>, Error> {
179 let Some(bytes) = self.get_optional(&format!("/tx/{txid}/raw"), SOURCE_MAX_BLOCK_BYTES)?
180 else {
181 return Ok(None);
182 };
183 let tx: Transaction = deserialize(&bytes).context("invalid source raw transaction")?;
184 ensure!(tx.compute_txid() == txid, "source transaction ID mismatch");
185 Ok(Some((tx, bytes)))
186 }
187
188 pub fn transaction_bundle(&self, txids: &[Txid]) -> Result<(Vec<u8>, Value), Error> {
189 ensure!(
190 !txids.is_empty() && txids.len() <= Limits::RECORDS,
191 "transaction fetch requires 1..={} transaction IDs",
192 Limits::RECORDS
193 );
194 let mut records = Vec::new();
195 let mut observations = Vec::new();
196 let mut fetched_bytes = 0usize;
197 let mut seen = std::collections::BTreeSet::new();
198 for &txid in txids {
199 ensure!(seen.insert(txid), "duplicate transaction ID");
200 let (tx, raw) = self.transaction(txid)?;
201 fetched_bytes = fetched_bytes
202 .checked_add(raw.len())
203 .context("fetch byte overflow")?;
204 ensure!(
205 fetched_bytes <= 8_000_000,
206 "transaction fetch exceeds byte limit"
207 );
208 let status: TxStatus = self.json(&format!("/tx/{txid}/status"))?;
209 status.validate()?;
210 for input in &tx.input {
211 if envelope::is_candidate(&input.witness) {
212 envelope::extract_reveal(&tx)?;
213 let (commit, commit_raw) =
214 self.transaction(tx.input[0].previous_output.txid)?;
215 fetched_bytes = fetched_bytes
216 .checked_add(commit_raw.len())
217 .context("fetch byte overflow")?;
218 ensure!(
219 fetched_bytes <= 8_000_000,
220 "transaction fetch exceeds byte limit"
221 );
222 let record = envelope::verify_reveal(&tx, &commit)?.record;
223 ensure!(
224 records.len() < Limits::RECORDS,
225 "bundle record limit reached"
226 );
227 container::inspect_header(&record)?;
228 records.push(record);
229 }
230 }
231 observations.push(json!({
232 "txid": txid.to_string(), "wtxid": tx.compute_wtxid().to_string(),
233 "raw_bytes": raw.len(),
234 "state": if status.confirmed { "provider_reported_confirmed" } else { "provider_reported_pending" },
235 "block_height": status.block_height,
236 "block_hash": status.block_hash.map(|h| h.to_string())
237 }));
238 }
239 ensure!(
240 !records.is_empty(),
241 "no private URMA records in fetched transactions"
242 );
243 let bundle = container::pack(&records)?;
244 let report = json!({
245 "status": "fetched", "network": self.network.to_string(), "provider": self.url,
246 "transactions": observations, "records": records.len(), "fetched_bytes": fetched_bytes,
247 "bundle_bytes": bundle.len(), "chain_inclusion_verified": false,
248 "content_authenticated": false, "evidence": "provider_transaction_bytes",
249 "note": "Transient provider availability, not atomic/global mempool membership or block inclusion. Authenticate/decrypt the bundle separately; no key or sender journal was used to fetch it."
250 });
251 Ok((bundle, report))
252 }
253
254 fn observe_transaction(
255 &self,
256 tx: &Transaction,
257 raw: &[u8],
258 tip_height: u64,
259 ) -> Result<Value, Error> {
260 let txid = tx.compute_txid();
261 let Some((found, found_raw)) = self.transaction_optional(txid)? else {
262 return Ok(json!({"txid": txid.to_string(), "state": "unknown", "confirmations": 0}));
263 };
264 ensure!(
265 found == *tx && found_raw == raw,
266 "source transaction witness differs from prepared plan"
267 );
268 let status: TxStatus = self.json(&format!("/tx/{txid}/status"))?;
269 let confirmations = self.verify_confirmation(&found, &status, tip_height)?;
270 Ok(
271 json!({"txid": txid.to_string(), "state": if status.confirmed {"confirmed"} else {"pending"},
272 "confirmations": confirmations, "block_height": status.block_height, "block_hash": status.block_hash.map(|h| h.to_string())}),
273 )
274 }
275
276 fn verify_confirmation(
277 &self,
278 tx: &Transaction,
279 status: &TxStatus,
280 tip: u64,
281 ) -> Result<u64, Error> {
282 if !status.confirmed {
283 status.validate()?;
284 return Ok(0);
285 }
286 let height = status
287 .block_height
288 .context("confirmed source status lacks block height")?;
289 let hash = status
290 .block_hash
291 .context("confirmed source status lacks block hash")?;
292 ensure!(height <= tip, "confirmation is beyond source tip snapshot");
293 ensure!(
294 self.block_hash(height)? == hash,
295 "source confirmation block is not active"
296 );
297 let (block, _) = self.block(hash)?;
298 ensure!(
299 block.txdata.iter().any(|candidate| candidate == tx),
300 "source transaction or witness absent from claimed block"
301 );
302 Ok(tip - height + 1)
303 }
304
305 pub fn funding(
306 &self,
307 address: &str,
308 allow_unconfirmed: bool,
309 ) -> Result<(Funding, Value), Error> {
310 let address = Address::from_str(address)?.require_network(self.network)?;
311 ensure!(
312 address.script_pubkey().is_p2wpkh(),
313 "funding requires a native P2WPKH address"
314 );
315 let tip_height = self.height()?;
316 let tip_hash = self.block_hash(tip_height)?;
317 let mut utxos: Vec<Utxo> = self.json(&format!("/address/{address}/utxo"))?;
318 ensure!(utxos.len() <= SOURCE_MAX_UTXOS, "source UTXO limit reached");
319 utxos.retain(|u| u.status.confirmed || allow_unconfirmed);
320 utxos.sort_by_key(|u| std::cmp::Reverse(u.value));
321 let utxo = utxos.first().context(
322 "no eligible source UTXO; pending funding requires explicit prepare-only opt-in",
323 )?;
324 let (tx, raw) = self.transaction(utxo.txid)?;
325 ensure!(
326 !tx.is_coinbase(),
327 "coinbase funding is unsupported; use a regular faucet output"
328 );
329 let output = tx
330 .output
331 .get(usize::try_from(utxo.vout)?)
332 .context("source funding vout is absent")?;
333 ensure!(
334 output.script_pubkey == address.script_pubkey(),
335 "source funding address/script mismatch"
336 );
337 ensure!(
338 output.value.to_sat() == utxo.value && utxo.value > 0,
339 "source funding amount mismatch"
340 );
341 let status: TxStatus = self.json(&format!("/tx/{}/status", utxo.txid))?;
342 ensure!(
343 status.confirmed == utxo.status.confirmed
344 && status.block_hash == utxo.status.block_hash
345 && status.block_height == utxo.status.block_height,
346 "source funding status changed; retry"
347 );
348 let confirmations = self.verify_confirmation(&tx, &status, tip_height)?;
349 let outspend: Value = self.json(&format!("/tx/{}/outspend/{}", utxo.txid, utxo.vout))?;
350 ensure!(
351 outspend["spent"].as_bool() == Some(false),
352 "source funding output is spent or status is missing"
353 );
354 ensure!(
355 self.block_hash(tip_height)? == tip_hash,
356 "source chain changed during funding lookup; retry"
357 );
358 let report = json!({
359 "network": self.network.to_string(), "provider": self.url, "genesis": self.genesis.to_string(),
360 "status": if status.confirmed { "confirmed" } else { "pending" }, "confirmed": status.confirmed,
361 "txid": utxo.txid.to_string(), "vout": utxo.vout, "outpoint": format!("{}:{}", utxo.txid, utxo.vout),
362 "value_sats": utxo.value, "height": status.block_height, "confirmations": confirmations,
363 "tip_height": tip_height, "tip_hash": tip_hash.to_string(), "start_height": tip_height,
364 "recommended_start_height": tip_height, "prepare_only": !status.confirmed,
365 "verification": if status.confirmed {
366 "raw transaction and confirmed block contents verified; provider trusted for chain selection and unspentness; header-chain difficulty transitions are not independently validated"
367 } else {
368 "raw transaction verified; provider reports pending and unspent; no confirmed block inclusion verified; provider trusted for chain selection and unspentness"
369 }
370 });
371 Ok((
372 Funding {
373 raw_transaction: hex::encode(raw),
374 vout: utxo.vout,
375 },
376 report,
377 ))
378 }
379
380 pub fn scan(&self, key: &[u8; 32], start_height: u64) -> Result<Scan, Error> {
381 Scan::from_recovery(backend::recover(
382 &ExplorerRecords {
383 source: self,
384 start_height,
385 },
386 key,
387 )?)
388 }
389
390 fn read_records(
391 &self,
392 start_height: u64,
393 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
394 ) -> Result<Observation, Error> {
395 self.read_range(
396 start_height,
397 u64::MAX,
398 &mut |height| {
399 tracing::trace!(height, "scan progress");
400 Ok(())
401 },
402 emit,
403 )
404 }
405
406 pub fn read_range(
407 &self,
408 start_height: u64,
409 tip_height: u64,
410 progress: &mut dyn FnMut(u64) -> Result<(), Error>,
411 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
412 ) -> Result<Observation, Error> {
413 let current = self.height()?;
414 let tip_height = if tip_height == u64::MAX {
415 current
416 } else {
417 tip_height
418 };
419 ensure!(tip_height <= current, "end height beyond source tip");
420 ensure!(
421 start_height <= tip_height,
422 "start height is beyond source tip"
423 );
424 ensure!(
425 tip_height - start_height < SOURCE_MAX_SCAN_BLOCKS,
426 "scan block limit reached; narrow starting height"
427 );
428 let tip_hash = self.block_hash(tip_height)?;
429 let mut previous = if start_height == 0 {
430 BlockHash::all_zeros()
431 } else {
432 self.block_hash(start_height - 1)?
433 };
434 let mut scanned_bytes = 0usize;
435 for height in start_height..=tip_height {
436 progress(height)?;
437 let hash = self.block_hash(height)?;
438 let (block, length) = self
439 .block(hash)
440 .with_context(|| format!("source block {height}"))?;
441 scanned_bytes = scanned_bytes
442 .checked_add(length)
443 .context("scan byte overflow")?;
444 ensure!(
445 scanned_bytes <= SOURCE_MAX_SCAN_BYTES,
446 "scan byte limit reached; narrow starting height"
447 );
448 ensure!(
449 block.header.prev_blockhash == previous,
450 "source chain changed during scan"
451 );
452 previous = hash;
453 self.emit_verified_records(&block, emit)?;
454 }
455 ensure!(
456 previous == tip_hash && self.block_hash(tip_height)? == tip_hash,
457 "source chain changed during scan; retry recovery"
458 );
459 Ok(Observation {
460 backend: Family::Bitcoin,
461 locator: Locator::BlockRange {
462 network: self.network.to_string(),
463 start_height,
464 tip_height,
465 tip_hash: tip_hash.to_string(),
466 },
467 evidence: Evidence::CommitmentsAndProviderChain,
468 scanned_bytes,
469 })
470 }
471
472 pub fn status_plan(&self, plan: &Plan) -> Result<Value, Error> {
473 let mut report = journal::validate_plan(plan)?;
474 ensure!(
475 plan.network == self.network.to_string(),
476 "source network differs from plan network"
477 );
478 let tip_height = self.height()?;
479 let tip_hash = self.block_hash(tip_height)?;
480 let mut statuses = Vec::new();
481 for encoded in std::iter::once(&plan.commit).chain(plan.reveals.iter()) {
482 let raw = hex::decode(encoded)?;
483 let tx: Transaction = deserialize(&raw)?;
484 statuses.push(self.observe_transaction(&tx, &raw, tip_height)?);
485 }
486 ensure!(
487 self.block_hash(tip_height)? == tip_hash,
488 "source chain changed during status lookup; retry"
489 );
490 let commit = &statuses[0];
491 let reveals = &statuses[1..];
492 let status = if commit["state"] == "unknown" {
493 ensure!(
494 reveals.iter().all(|s| s["state"] == "unknown"),
495 "source knows reveal while commit is absent"
496 );
497 "prepared"
498 } else if commit["state"] == "pending" {
499 ensure!(
500 reveals.iter().all(|s| s["state"] != "confirmed"),
501 "source confirms reveal while commit is pending"
502 );
503 "commit_pending"
504 } else if reveals.iter().all(|s| s["state"] == "confirmed") {
505 "confirmed"
506 } else if reveals.iter().any(|s| s["state"] == "unknown") {
507 "awaiting_reveals"
508 } else {
509 "confirming"
510 };
511 report["status"] = json!(status);
512 report["provider"] = json!(self.url);
513 report["tip_height"] = json!(tip_height);
514 report["tip_hash"] = json!(tip_hash.to_string());
515 report["commit_status"] = commit.clone();
516 report["reveal_statuses"] = json!(reveals);
517 let known_count = statuses.iter().filter(|s| s["state"] != "unknown").count();
518 let confirmed_count = statuses
519 .iter()
520 .filter(|s| s["state"] == "confirmed")
521 .count();
522 report["verified_transaction_count"] = json!(known_count);
523 report["verified_confirmed_transaction_count"] = json!(confirmed_count);
524 report["verification"] = json!(format!(
525 "{known_count} source transactions matched prepared bytes and witnesses; {confirmed_count} confirmed transactions verified inside blocks; provider trusted for chain selection; header-chain difficulty transitions are not independently validated"
526 ));
527 Ok(report)
528 }
529}
530
531pub struct ExplorerRecords<'a> {
532 pub source: &'a Source,
533 pub start_height: u64,
534}
535
536impl RecordSource for ExplorerRecords<'_> {
537 fn read_records(
538 &self,
539 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
540 ) -> Result<Observation, Error> {
541 self.source.read_records(self.start_height, emit)
542 }
543}
544
545fn checked_url(input: &str) -> Result<String, Error> {
546 let url = url::Url::parse(input).context("invalid source URL")?;
547 match (url.password(), url.query(), url.fragment()) {
548 (None, None, None) => {}
549 components => {
550 tracing::warn!(
551 component_count = [components.0, components.1, components.2].iter().count(),
552 "forbidden URL components"
553 );
554 bail!("URL credentials, query and fragment are forbidden");
555 }
556 }
557 ensure!(
558 url.username().is_empty(),
559 "source URL cannot contain credentials, query, or fragment"
560 );
561 let local = match url.host() {
562 Some(url::Host::Domain("localhost")) => true,
563 Some(url::Host::Ipv4(ip)) => ip.is_loopback(),
564 Some(url::Host::Ipv6(ip)) => ip.is_loopback(),
565 Some(url::Host::Domain(domain)) => domain == "localhost",
566 None => return Err(Error::Missing("source URL has no host".into())),
567 };
568 match url.scheme() {
569 "https" => (),
570 "http" if local => (),
571 _ => bail!("source requires HTTPS; HTTP is allowed only on localhost"),
572 }
573
574 Ok(url.as_str().trim_end_matches('/').to_owned())
575}
576
577impl TxStatus {
578 fn validate(&self) -> Result<(), Error> {
579 match (&self.block_hash, self.block_height) {
580 (Some(hash), Some(height)) => ensure!(
581 self.confirmed,
582 "unconfirmed transaction includes block {hash} at {height}"
583 ),
584 (None, None) => ensure!(
585 !self.confirmed,
586 "confirmed transaction lacks block location"
587 ),
588 (Some(hash), None) => bail!("block {hash} lacks height"),
589 (None, Some(height)) => bail!("block at height {height} lacks hash"),
590 }
591 Ok(())
592 }
593}
594
595impl Source {
596 fn emit_verified_records(
597 &self,
598 block: &Block,
599 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
600 ) -> Result<(), Error> {
601 for tx in &block.txdata {
602 if !tx
603 .input
604 .iter()
605 .any(|input| envelope::is_candidate(&input.witness))
606 {
607 continue;
608 }
609 match envelope::extract_reveal(tx) {
610 Ok(parsed) => {
611 tracing::trace!(author=%parsed.author, "candidate with canonical transaction shape")
612 }
613 Err(error) => {
614 tracing::warn!(%error, "invalid reveal rejected");
615 continue;
616 }
617 }
618 let (commit, raw) = self.transaction(tx.input[0].previous_output.txid)?;
619 ensure!(
620 raw.len() <= SOURCE_MAX_BLOCK_BYTES,
621 "commit proof exceeds capacity"
622 );
623 match envelope::verify_reveal(tx, &commit) {
624 Ok(parsed) => emit(&parsed.record)?,
625 Err(error) => tracing::warn!(%error, "invalid author proof rejected"),
626 }
627 }
628 Ok(())
629 }
630}