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: Block = deserialize(&bytes).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 ensure!(utxos.len() <= SOURCE_MAX_UTXOS, "source UTXO limit reached");
318 utxos.retain(|u| u.status.confirmed || allow_unconfirmed);
319 utxos.sort_by_key(|u| std::cmp::Reverse(u.value));
320 let utxo = utxos.first().context(
321 "no eligible source UTXO; pending funding requires explicit prepare-only opt-in",
322 )?;
323 let (tx, raw) = self.transaction(utxo.txid)?;
324 ensure!(
325 !tx.is_coinbase(),
326 "coinbase funding is unsupported; use a regular faucet output"
327 );
328 let output = tx
329 .output
330 .get(usize::try_from(utxo.vout)?)
331 .context("source funding vout is absent")?;
332 ensure!(
333 output.script_pubkey == address.script_pubkey(),
334 "source funding address/script mismatch"
335 );
336 ensure!(
337 output.value.to_sat() == utxo.value && utxo.value > 0,
338 "source funding amount mismatch"
339 );
340 let status: TxStatus = self.json(&format!("/tx/{}/status", utxo.txid))?;
341 ensure!(
342 status.confirmed == utxo.status.confirmed
343 && status.block_hash == utxo.status.block_hash
344 && status.block_height == utxo.status.block_height,
345 "source funding status changed; retry"
346 );
347 let confirmations = self.verify_confirmation(&tx, &status, tip_height)?;
348 let outspend: Value = self.json(&format!("/tx/{}/outspend/{}", utxo.txid, utxo.vout))?;
349 ensure!(
350 outspend["spent"].as_bool() == Some(false),
351 "source funding output is spent or status is missing"
352 );
353 ensure!(
354 self.block_hash(tip_height)? == tip_hash,
355 "source chain changed during funding lookup; retry"
356 );
357 let report = json!({
358 "network": self.network.to_string(), "provider": self.url, "genesis": self.genesis.to_string(),
359 "status": if status.confirmed { "confirmed" } else { "pending" }, "confirmed": status.confirmed,
360 "txid": utxo.txid.to_string(), "vout": utxo.vout, "outpoint": format!("{}:{}", utxo.txid, utxo.vout),
361 "value_sats": utxo.value, "height": status.block_height, "confirmations": confirmations,
362 "tip_height": tip_height, "tip_hash": tip_hash.to_string(), "start_height": tip_height,
363 "recommended_start_height": tip_height, "prepare_only": !status.confirmed,
364 "verification": if status.confirmed {
365 "raw transaction and confirmed block contents verified; provider trusted for chain selection and unspentness; header-chain difficulty transitions are not independently validated"
366 } else {
367 "raw transaction verified; provider reports pending and unspent; no confirmed block inclusion verified; provider trusted for chain selection and unspentness"
368 }
369 });
370 Ok((
371 Funding {
372 raw_transaction: hex::encode(raw),
373 vout: utxo.vout,
374 },
375 report,
376 ))
377 }
378
379 pub fn scan(&self, key: &[u8; 32], start_height: u64) -> Result<Scan, Error> {
380 Scan::from_recovery(backend::recover(
381 &ExplorerRecords {
382 source: self,
383 start_height,
384 },
385 key,
386 )?)
387 }
388
389 fn read_records(
390 &self,
391 start_height: u64,
392 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
393 ) -> Result<Observation, Error> {
394 self.read_range(
395 start_height,
396 u64::MAX,
397 &mut |height| {
398 tracing::trace!(height, "scan progress");
399 Ok(())
400 },
401 emit,
402 )
403 }
404
405 pub fn read_range(
406 &self,
407 start_height: u64,
408 tip_height: u64,
409 progress: &mut dyn FnMut(u64) -> Result<(), Error>,
410 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
411 ) -> Result<Observation, Error> {
412 let current = self.height()?;
413 let tip_height = if tip_height == u64::MAX {
414 current
415 } else {
416 tip_height
417 };
418 ensure!(tip_height <= current, "end height beyond source tip");
419 ensure!(
420 start_height <= tip_height,
421 "start height is beyond source tip"
422 );
423 ensure!(
424 tip_height - start_height < SOURCE_MAX_SCAN_BLOCKS,
425 "scan block limit reached; narrow starting height"
426 );
427 let tip_hash = self.block_hash(tip_height)?;
428 let mut previous = if start_height == 0 {
429 BlockHash::all_zeros()
430 } else {
431 self.block_hash(start_height - 1)?
432 };
433 let mut scanned_bytes = 0usize;
434 for height in start_height..=tip_height {
435 progress(height)?;
436 let hash = self.block_hash(height)?;
437 let (block, length) = self
438 .block(hash)
439 .with_context(|| format!("source block {height}"))?;
440 scanned_bytes = scanned_bytes
441 .checked_add(length)
442 .context("scan byte overflow")?;
443 ensure!(
444 scanned_bytes <= SOURCE_MAX_SCAN_BYTES,
445 "scan byte limit reached; narrow starting height"
446 );
447 ensure!(
448 block.header.prev_blockhash == previous,
449 "source chain changed during scan"
450 );
451 previous = hash;
452 self.emit_verified_records(&block, emit)?;
453 }
454 ensure!(
455 previous == tip_hash && self.block_hash(tip_height)? == tip_hash,
456 "source chain changed during scan; retry recovery"
457 );
458 Ok(Observation {
459 backend: Family::Bitcoin,
460 locator: Locator::BlockRange {
461 network: self.network.to_string(),
462 start_height,
463 tip_height,
464 tip_hash: tip_hash.to_string(),
465 },
466 evidence: Evidence::CommitmentsAndProviderChain,
467 scanned_bytes,
468 })
469 }
470
471 pub fn status_plan(&self, plan: &Plan) -> Result<Value, Error> {
472 let mut report = journal::validate_plan(plan)?;
473 ensure!(
474 plan.network == self.network.to_string(),
475 "source network differs from plan network"
476 );
477 let tip_height = self.height()?;
478 let tip_hash = self.block_hash(tip_height)?;
479 let mut statuses = Vec::new();
480 for encoded in std::iter::once(&plan.commit).chain(plan.reveals.iter()) {
481 let raw = hex::decode(encoded)?;
482 let tx: Transaction = deserialize(&raw)?;
483 statuses.push(self.observe_transaction(&tx, &raw, tip_height)?);
484 }
485 ensure!(
486 self.block_hash(tip_height)? == tip_hash,
487 "source chain changed during status lookup; retry"
488 );
489 let commit = &statuses[0];
490 let reveals = &statuses[1..];
491 let status = if commit["state"] == "unknown" {
492 ensure!(
493 reveals.iter().all(|s| s["state"] == "unknown"),
494 "source knows reveal while commit is absent"
495 );
496 "prepared"
497 } else if commit["state"] == "pending" {
498 ensure!(
499 reveals.iter().all(|s| s["state"] != "confirmed"),
500 "source confirms reveal while commit is pending"
501 );
502 "commit_pending"
503 } else if reveals.iter().all(|s| s["state"] == "confirmed") {
504 "confirmed"
505 } else if reveals.iter().any(|s| s["state"] == "unknown") {
506 "awaiting_reveals"
507 } else {
508 "confirming"
509 };
510 report["status"] = json!(status);
511 report["provider"] = json!(self.url);
512 report["tip_height"] = json!(tip_height);
513 report["tip_hash"] = json!(tip_hash.to_string());
514 report["commit_status"] = commit.clone();
515 report["reveal_statuses"] = json!(reveals);
516 let known_count = statuses.iter().filter(|s| s["state"] != "unknown").count();
517 let confirmed_count = statuses
518 .iter()
519 .filter(|s| s["state"] == "confirmed")
520 .count();
521 report["verified_transaction_count"] = json!(known_count);
522 report["verified_confirmed_transaction_count"] = json!(confirmed_count);
523 report["verification"] = json!(format!(
524 "{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"
525 ));
526 Ok(report)
527 }
528}
529
530pub struct ExplorerRecords<'a> {
531 pub source: &'a Source,
532 pub start_height: u64,
533}
534
535impl RecordSource for ExplorerRecords<'_> {
536 fn read_records(
537 &self,
538 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
539 ) -> Result<Observation, Error> {
540 self.source.read_records(self.start_height, emit)
541 }
542}
543
544fn checked_url(input: &str) -> Result<String, Error> {
545 let url = url::Url::parse(input).context("invalid source URL")?;
546 match (url.password(), url.query(), url.fragment()) {
547 (None, None, None) => {}
548 components => {
549 tracing::warn!(
550 component_count = [components.0, components.1, components.2].iter().count(),
551 "forbidden URL components"
552 );
553 bail!("URL credentials, query and fragment are forbidden");
554 }
555 }
556 ensure!(
557 url.username().is_empty(),
558 "source URL cannot contain credentials, query, or fragment"
559 );
560 let local = match url.host() {
561 Some(url::Host::Domain("localhost")) => true,
562 Some(url::Host::Ipv4(ip)) => ip.is_loopback(),
563 Some(url::Host::Ipv6(ip)) => ip.is_loopback(),
564 Some(url::Host::Domain(domain)) => domain == "localhost",
565 None => return Err(Error::Missing("source URL has no host".into())),
566 };
567 match url.scheme() {
568 "https" => (),
569 "http" if local => (),
570 _ => bail!("source requires HTTPS; HTTP is allowed only on localhost"),
571 }
572
573 Ok(url.as_str().trim_end_matches('/').to_owned())
574}
575
576impl TxStatus {
577 fn validate(&self) -> Result<(), Error> {
578 match (&self.block_hash, self.block_height) {
579 (Some(hash), Some(height)) => ensure!(
580 self.confirmed,
581 "unconfirmed transaction includes block {hash} at {height}"
582 ),
583 (None, None) => ensure!(
584 !self.confirmed,
585 "confirmed transaction lacks block location"
586 ),
587 (Some(hash), None) => bail!("block {hash} lacks height"),
588 (None, Some(height)) => bail!("block at height {height} lacks hash"),
589 }
590 Ok(())
591 }
592}
593
594impl Source {
595 fn emit_verified_records(
596 &self,
597 block: &Block,
598 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
599 ) -> Result<(), Error> {
600 for tx in &block.txdata {
601 if !tx
602 .input
603 .iter()
604 .any(|input| envelope::is_candidate(&input.witness))
605 {
606 continue;
607 }
608 match envelope::extract_reveal(tx) {
609 Ok(parsed) => {
610 tracing::trace!(author=%parsed.author, "candidate with canonical transaction shape")
611 }
612 Err(error) => {
613 tracing::warn!(%error, "invalid reveal rejected");
614 continue;
615 }
616 }
617 let (commit, raw) = self.transaction(tx.input[0].previous_output.txid)?;
618 ensure!(
619 raw.len() <= SOURCE_MAX_BLOCK_BYTES,
620 "commit proof exceeds capacity"
621 );
622 match envelope::verify_reveal(tx, &commit) {
623 Ok(parsed) => emit(&parsed.record)?,
624 Err(error) => tracing::warn!(%error, "invalid author proof rejected"),
625 }
626 }
627 Ok(())
628 }
629}