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