From d9b3b98512c951780109992db6a562991258c32a Mon Sep 17 00:00:00 2001 From: Tommy Volk Date: Fri, 28 Aug 2026 10:59:36 -0500 Subject: [PATCH] feat(rfq): add authenticated provider protocol --- Cargo.lock | 30 + Cargo.toml | 4 + crates/deadcat-rfq-iroh/Cargo.toml | 27 + crates/deadcat-rfq-iroh/src/client.rs | 616 ++++++ crates/deadcat-rfq-iroh/src/handler.rs | 77 + crates/deadcat-rfq-iroh/src/lib.rs | 18 + crates/deadcat-rfq-iroh/src/server.rs | 413 ++++ crates/deadcat-rfq-iroh/tests/round_trip.rs | 233 +++ crates/deadcat-rfq-provider/src/lib.rs | 14 +- crates/deadcat-rfq-provider/src/store.rs | 72 +- .../deadcat-rfq-provider/src/store/tests.rs | 74 + crates/deadcat-rfq-rpc/Cargo.toml | 23 + crates/deadcat-rfq-rpc/src/codec.rs | 179 ++ crates/deadcat-rfq-rpc/src/lib.rs | 464 +++++ crates/deadcat-rfq-rpc/src/quote.rs | 1836 +++++++++++++++++ justfile | 4 +- 16 files changed, 4075 insertions(+), 9 deletions(-) create mode 100644 crates/deadcat-rfq-iroh/Cargo.toml create mode 100644 crates/deadcat-rfq-iroh/src/client.rs create mode 100644 crates/deadcat-rfq-iroh/src/handler.rs create mode 100644 crates/deadcat-rfq-iroh/src/lib.rs create mode 100644 crates/deadcat-rfq-iroh/src/server.rs create mode 100644 crates/deadcat-rfq-iroh/tests/round_trip.rs create mode 100644 crates/deadcat-rfq-rpc/Cargo.toml create mode 100644 crates/deadcat-rfq-rpc/src/codec.rs create mode 100644 crates/deadcat-rfq-rpc/src/lib.rs create mode 100644 crates/deadcat-rfq-rpc/src/quote.rs diff --git a/Cargo.lock b/Cargo.lock index 605cbb5..b8d7f59 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1034,6 +1034,20 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "deadcat-rfq-iroh" +version = "0.1.0-alpha" +dependencies = [ + "deadcat-iroh", + "deadcat-rfq-rpc", + "deadcat-types", + "elements", + "iroh", + "thiserror 2.0.18", + "tokio", + "tracing", +] + [[package]] name = "deadcat-rfq-provider" version = "0.1.0-alpha" @@ -1049,6 +1063,22 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "deadcat-rfq-rpc" +version = "0.1.0-alpha" +dependencies = [ + "deadcat-types", + "elements", + "hex", + "iroh", + "postcard", + "rand 0.8.7", + "serde", + "serde_json", + "sha2 0.10.9", + "thiserror 2.0.18", +] + [[package]] name = "deadcat-rfq-wallet" version = "0.1.0-alpha" diff --git a/Cargo.toml b/Cargo.toml index 8055ca0..9ae7bd4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,8 @@ members = [ "crates/deadcat-node", "crates/deadcat-cli", "crates/deadcat-rfq-provider", + "crates/deadcat-rfq-iroh", + "crates/deadcat-rfq-rpc", "crates/deadcat-rfq-wallet", "crates/deadcat-rfq", ] @@ -53,6 +55,8 @@ deadcat-client = { path = "crates/deadcat-client" } deadcat-rpc = { path = "crates/deadcat-rpc" } deadcat-iroh = { path = "crates/deadcat-iroh" } deadcat-rfq-provider = { path = "crates/deadcat-rfq-provider" } +deadcat-rfq-iroh = { path = "crates/deadcat-rfq-iroh" } +deadcat-rfq-rpc = { path = "crates/deadcat-rfq-rpc" } deadcat-rfq-wallet = { path = "crates/deadcat-rfq-wallet" } deadcat-rfq = { path = "crates/deadcat-rfq" } diff --git a/crates/deadcat-rfq-iroh/Cargo.toml b/crates/deadcat-rfq-iroh/Cargo.toml new file mode 100644 index 0000000..58be865 --- /dev/null +++ b/crates/deadcat-rfq-iroh/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "deadcat-rfq-iroh" +description = "Bounded authenticated Iroh transport for the Deadcat RFQ protocol." +version.workspace = true +edition.workspace = true +publish.workspace = true + +[lints] +workspace = true + +[dependencies] +deadcat-iroh.workspace = true +deadcat-rfq-rpc.workspace = true +iroh.workspace = true +thiserror.workspace = true +tracing.workspace = true + +[target.'cfg(all(target_family = "wasm", target_os = "unknown"))'.dependencies] +tokio = { version = "1", default-features = false, features = ["io-util", "rt", "sync", "time"] } + +[target.'cfg(not(all(target_family = "wasm", target_os = "unknown")))'.dependencies] +tokio.workspace = true + +[dev-dependencies] +deadcat-types.workspace = true +elements.workspace = true +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/deadcat-rfq-iroh/src/client.rs b/crates/deadcat-rfq-iroh/src/client.rs new file mode 100644 index 0000000..3090d8b --- /dev/null +++ b/crates/deadcat-rfq-iroh/src/client.rs @@ -0,0 +1,616 @@ +//! Persistent-identity native Iroh client for Deadcat RFQ RPC. + +use std::sync::Arc; +use std::time::Duration; + +use deadcat_iroh::wire::{ + DEFAULT_INBOUND_BUDGET_BYTES, InboundBudget, MAX_FRAME_BYTES, WireError, read_message, + write_message, +}; +use deadcat_rfq_rpc::{ + FirmQuoteValidationError, Request, RequestEnvelope, RequestId, Response, RpcError, RpcOutcome, + SCHEMA_VERSION, ServerEnvelope, +}; +use iroh::endpoint::{Connection, presets}; +use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; + +use crate::ALPN; + +#[derive(Clone, Debug)] +pub struct ClientConfig { + pub max_in_flight_requests: usize, + pub inbound_budget_bytes: usize, + pub connect_timeout: Duration, + pub request_timeout: Duration, +} + +impl Default for ClientConfig { + fn default() -> Self { + Self { + max_in_flight_requests: 32, + inbound_budget_bytes: DEFAULT_INBOUND_BUDGET_BYTES, + connect_timeout: Duration::from_secs(20), + request_timeout: Duration::from_secs(30), + } + } +} + +impl ClientConfig { + fn validate(&self) -> Result<(), ClientError> { + if self.max_in_flight_requests == 0 { + return Err(ClientError::InvalidConfig( + "max_in_flight_requests must be nonzero", + )); + } + if self.inbound_budget_bytes < MAX_FRAME_BYTES + || self.inbound_budget_bytes > usize::try_from(u32::MAX).expect("u32 fits usize") + { + return Err(ClientError::InvalidConfig( + "inbound_budget_bytes must fit at least one maximum frame and be <= u32::MAX", + )); + } + if self.connect_timeout == Duration::ZERO || self.request_timeout == Duration::ZERO { + return Err(ClientError::InvalidConfig("timeouts must be nonzero")); + } + Ok(()) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum ClientError { + #[error("invalid client configuration: {0}")] + InvalidConfig(&'static str), + #[error("Iroh connect error: {0}")] + Connect(#[from] iroh::endpoint::ConnectError), + #[error("Iroh connection error: {0}")] + Connection(#[from] iroh::endpoint::ConnectionError), + #[error("wire error: {0}")] + Wire(#[from] WireError), + #[error("server returned RFQ error: {0:?}")] + Rpc(RpcError), + #[error("invalid RFQ request: {0}")] + InvalidRequest(#[source] FirmQuoteValidationError), + #[error("invalid RFQ response: {0}")] + InvalidResponse(#[source] FirmQuoteValidationError), + #[error("request timed out")] + Timeout, + #[error("response schema {actual} does not match expected schema {expected}")] + SchemaMismatch { expected: u32, actual: u32 }, + #[error("response request id {actual:?} does not match request id {expected:?}")] + RequestIdMismatch { + expected: RequestId, + actual: RequestId, + }, + #[error("response stream contained trailing data")] + TrailingData, + #[error("wrong response shape for request")] + WrongResponseShape, + #[error("response identifiers do not match the request")] + ResponseRequestMismatch, + #[error("response provider identity does not match the authenticated Iroh peer")] + ProviderIdentityMismatch, + #[error("Iroh endpoint error: {0}")] + Iroh(String), +} + +/// Connected RFQ client whose identity comes from a caller-owned persistent key. +pub struct Client { + endpoint: Endpoint, + connection: Connection, + config: ClientConfig, + inbound_budget: InboundBudget, + in_flight: Arc, +} + +impl Client { + /// Connect using normal Iroh discovery and relay configuration. + /// + /// There is intentionally no overload that generates an ephemeral key: + /// the authenticated endpoint ID is part of durable reservation ownership. + pub async fn connect( + target: impl Into, + secret_key: SecretKey, + config: ClientConfig, + ) -> Result { + config.validate()?; + let endpoint = Endpoint::builder(presets::N0) + .secret_key(secret_key) + .bind() + .await + .map_err(|error| ClientError::Iroh(error.to_string()))?; + Self::connect_endpoint(endpoint, target.into(), config).await + } + + /// Connect directly with relay and discovery disabled. + pub async fn dial_direct( + target: EndpointAddr, + secret_key: SecretKey, + config: ClientConfig, + ) -> Result { + config.validate()?; + let endpoint = Endpoint::builder(presets::N0) + .secret_key(secret_key) + .relay_mode(RelayMode::Disabled) + .bind() + .await + .map_err(|error| ClientError::Iroh(error.to_string()))?; + Self::connect_endpoint(endpoint, target, config).await + } + + async fn connect_endpoint( + endpoint: Endpoint, + target: EndpointAddr, + config: ClientConfig, + ) -> Result { + let connection = + tokio::time::timeout(config.connect_timeout, endpoint.connect(target, ALPN)) + .await + .map_err(|_| ClientError::Timeout)??; + Ok(Self { + endpoint, + connection, + inbound_budget: InboundBudget::new(config.inbound_budget_bytes), + in_flight: Arc::new(Semaphore::new(config.max_in_flight_requests)), + config, + }) + } + + /// Execute one low-level RFQ protocol request. + /// + /// Transport and structure validation do not make a returned + /// [`deadcat_rfq_rpc::SignedFirmQuote`] trusted trading authority. Call + /// `SignedFirmQuote::verify_at` with the authenticated endpoint IDs, + /// original request, idempotency key, and current time before using it. + pub async fn call(&self, envelope: RequestEnvelope) -> Result { + let deadline = tokio::time::Instant::now() + self.config.request_timeout; + envelope.validate_version().map_err(ClientError::Rpc)?; + envelope + .request + .validate() + .map_err(ClientError::InvalidRequest)?; + let permit = acquire_request_permit(Arc::clone(&self.in_flight), deadline).await?; + tokio::time::timeout_at(deadline, async { + let _permit = permit; + let (mut send, mut recv) = self.connection.open_bi().await?; + write_message(&mut send, &envelope).await?; + send.finish() + .map_err(|error| ClientError::Iroh(error.to_string()))?; + + let response: ServerEnvelope = read_message(&mut recv, &self.inbound_budget).await?; + let mut trailing = [0_u8; 1]; + if recv + .read(&mut trailing) + .await + .map_err(|error| ClientError::Iroh(error.to_string()))? + .is_some() + { + return Err(ClientError::TrailingData); + } + validate_response(&response, envelope.request_id)?; + let value = outcome_value(response.outcome)?; + validate_response_value(&envelope.request, &value)?; + validate_provider_identity(self.provider_endpoint_id(), &value)?; + Ok(value) + }) + .await + .map_err(|_| ClientError::Timeout)? + } + + #[must_use] + pub fn endpoint_id(&self) -> EndpointId { + self.endpoint.id() + } + + /// Authenticated endpoint identity of the connected RFQ provider. + /// + /// Callers use this value when verifying quote attestations, avoiding a + /// second identity source outside the transport connection. + #[must_use] + pub fn provider_endpoint_id(&self) -> EndpointId { + self.connection.remote_id() + } + + pub async fn close(self) { + self.connection.close(0_u32.into(), b"client closed"); + self.endpoint.close().await; + } +} + +async fn acquire_request_permit( + in_flight: Arc, + deadline: tokio::time::Instant, +) -> Result { + tokio::time::timeout_at(deadline, in_flight.acquire_owned()) + .await + .map_err(|_| ClientError::Timeout)? + .map_err(|_| ClientError::Iroh("client request semaphore closed".into())) +} + +fn outcome_value(outcome: RpcOutcome) -> Result { + match outcome { + RpcOutcome::Success { value } => Ok(value), + RpcOutcome::Error { error } => Err(ClientError::Rpc(error)), + } +} + +fn validate_response(response: &ServerEnvelope, request_id: RequestId) -> Result<(), ClientError> { + if response.schema_version != SCHEMA_VERSION { + return Err(ClientError::SchemaMismatch { + expected: SCHEMA_VERSION, + actual: response.schema_version, + }); + } + if response.request_id != request_id { + return Err(ClientError::RequestIdMismatch { + expected: request_id, + actual: response.request_id, + }); + } + Ok(()) +} + +fn validate_response_shape(request: &Request, response: &Response) -> Result<(), ClientError> { + if matches!( + (request, response), + (Request::GetInfo, Response::Info { .. }) + | (Request::RequestFirmQuote { .. }, Response::FirmQuote { .. }) + | ( + Request::CancelReservation { .. }, + Response::ReservationCancelled { .. } + ) + | (Request::BlindPset { .. }, Response::BlindedPset { .. }) + | (Request::Execute { .. }, Response::ExecutionAccepted { .. }) + | ( + Request::GetReservationStatus { .. }, + Response::ReservationStatus { .. } + ) + ) { + Ok(()) + } else { + Err(ClientError::WrongResponseShape) + } +} + +fn validate_response_value(request: &Request, response: &Response) -> Result<(), ClientError> { + validate_response_shape(request, response)?; + validate_response_request_binding(request, response)?; + response.validate().map_err(ClientError::InvalidResponse) +} + +fn validate_response_request_binding( + request: &Request, + response: &Response, +) -> Result<(), ClientError> { + let matches = match (request, response) { + ( + Request::RequestFirmQuote { + idempotency_key, + request, + }, + Response::FirmQuote { quote, .. }, + ) => firm_quote_binding_matches( + *idempotency_key, + request, + quote.attestation.idempotency_key, + "e.quote.request, + ), + ( + Request::CancelReservation { reservation_id }, + Response::ReservationCancelled { status }, + ) + | (Request::Execute { reservation_id, .. }, Response::ExecutionAccepted { status }) + | ( + Request::GetReservationStatus { reservation_id }, + Response::ReservationStatus { status }, + ) => status.reservation_id == *reservation_id, + ( + Request::BlindPset { reservation_id, .. }, + Response::BlindedPset { + reservation_id: returned, + .. + }, + ) => returned == reservation_id, + (Request::GetInfo, Response::Info { .. }) => true, + _ => false, + }; + if matches { + Ok(()) + } else { + Err(ClientError::ResponseRequestMismatch) + } +} + +fn firm_quote_binding_matches( + expected_idempotency_key: deadcat_rfq_rpc::IdempotencyKeyDto, + expected_request: &deadcat_rfq_rpc::FirmQuoteRequestDto, + actual_idempotency_key: deadcat_rfq_rpc::IdempotencyKeyDto, + actual_request: &deadcat_rfq_rpc::FirmQuoteRequestDto, +) -> bool { + actual_idempotency_key == expected_idempotency_key && actual_request == expected_request +} + +fn validate_provider_identity( + provider_endpoint: EndpointId, + response: &Response, +) -> Result<(), ClientError> { + let expected = *provider_endpoint.as_bytes(); + let matches = match response { + Response::Info { info } => info.provider_endpoint.to_bytes() == expected, + Response::FirmQuote { quote, .. } => { + quote.quote.provider_endpoint.to_bytes() == expected + && quote.attestation.provider_endpoint.to_bytes() == expected + } + Response::ReservationCancelled { .. } + | Response::BlindedPset { .. } + | Response::ExecutionAccepted { .. } + | Response::ReservationStatus { .. } => true, + }; + if matches { + Ok(()) + } else { + Err(ClientError::ProviderIdentityMismatch) + } +} + +#[cfg(test)] +mod tests { + use deadcat_rfq_rpc::{ + AssetAmountDto, FirmQuoteRequestDto, FixedBytes32, FixedBytes33, InputPlacementDto, + OutputPlacementDto, ProviderCapability, ProviderInfo, QuoteContextDto, QuoteKindDto, + QuoteRecipientDto, ReservationStateDto, ReservationStatusDto, RpcErrorCode, RpcOutcome, + SettlementLayoutDto, SettlementPset, + }; + use deadcat_types::{ContractId, LiquidNetwork}; + use elements::hashes::Hash as _; + use elements::pset::PartiallySignedTransaction; + use elements::secp256k1_zkp::{PublicKey, Secp256k1, SecretKey as SecpSecretKey}; + use elements::{AssetId, BlockHash, OutPoint, Txid}; + + use super::*; + + fn error_response(request_id: RequestId) -> ServerEnvelope { + ServerEnvelope::new( + request_id, + RpcOutcome::Error { + error: RpcError::new(RpcErrorCode::InternalError, "test") + .expect("bounded test error"), + }, + ) + } + + fn asset(marker: u8) -> AssetId { + AssetId::from_slice(&[marker; 32]).expect("test asset") + } + + fn quote_request(marker: u8) -> FirmQuoteRequestDto { + let blinding_secret = SecpSecretKey::from_slice(&[marker; 32]).expect("test secret"); + let blinding_public_key = PublicKey::from_secret_key(&Secp256k1::new(), &blinding_secret); + FirmQuoteRequestDto { + context: QuoteContextDto { + network: LiquidNetwork::ElementsRegtest, + genesis_hash: BlockHash::from_byte_array([marker.wrapping_add(1); 32]), + market: ContractId::new(OutPoint::new( + Txid::from_byte_array([marker.wrapping_add(2); 32]), + 0, + )), + policy_asset: asset(marker.wrapping_add(3)), + }, + kind: QuoteKindDto::ExactIn { + input: AssetAmountDto { + asset: asset(marker.wrapping_add(4)), + amount: 100, + }, + output_asset: asset(marker.wrapping_add(5)), + minimum_output: 90, + }, + recipient: QuoteRecipientDto { + script_pubkey: vec![0x51], + blinding_public_key: FixedBytes33::new(blinding_public_key.serialize()), + }, + maximum_input_asset_venue_fee: 10, + } + } + + fn status(reservation_id: FixedBytes32) -> ReservationStatusDto { + ReservationStatusDto { + reservation_id, + quote_commitment: FixedBytes32::new([0x77; 32]), + created_at_millis: 1, + accept_before_millis: 2, + state: ReservationStateDto::Reserved, + } + } + + #[test] + fn response_schema_and_request_id_must_match() { + let expected = RequestId(41); + let mut wrong_schema = error_response(expected); + wrong_schema.schema_version += 1; + assert!(matches!( + validate_response(&wrong_schema, expected), + Err(ClientError::SchemaMismatch { .. }) + )); + + let wrong_id = error_response(RequestId(42)); + assert!(matches!( + validate_response(&wrong_id, expected), + Err(ClientError::RequestIdMismatch { .. }) + )); + } + + #[test] + fn response_variant_must_match_request_method() { + let request = Request::CancelReservation { + reservation_id: FixedBytes32::new([0x11; 32]), + }; + let response = Response::Info { + info: ProviderInfo { + provider_endpoint: FixedBytes32::new([0x22; 32]), + network: LiquidNetwork::ElementsRegtest, + genesis_hash: "0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .expect("test block hash"), + policy_asset: "0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .expect("test asset id"), + capabilities: vec![ProviderCapability::FirmQuotes], + }, + }; + assert!(matches!( + validate_response_shape(&request, &response), + Err(ClientError::WrongResponseShape) + )); + } + + #[test] + fn method_matching_response_must_pass_semantic_validation() { + let reservation_id = FixedBytes32::new([0x33; 32]); + let request = Request::GetReservationStatus { reservation_id }; + let response = Response::ReservationStatus { + status: ReservationStatusDto { + reservation_id, + quote_commitment: FixedBytes32::new([0x44; 32]), + created_at_millis: 2, + accept_before_millis: 1, + state: ReservationStateDto::Reserved, + }, + }; + assert!(matches!( + validate_response_value(&request, &response), + Err(ClientError::InvalidResponse(_)) + )); + } + + #[test] + fn response_provider_must_match_authenticated_connection_peer() { + let provider_key = SecretKey::generate(); + let wrong_provider = Response::Info { + info: ProviderInfo { + provider_endpoint: FixedBytes32::new([0x66; 32]), + network: LiquidNetwork::ElementsRegtest, + genesis_hash: "0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .expect("test block hash"), + policy_asset: "0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .expect("test asset id"), + capabilities: vec![ProviderCapability::FirmQuotes], + }, + }; + assert!(matches!( + validate_provider_identity(provider_key.public(), &wrong_provider), + Err(ClientError::ProviderIdentityMismatch) + )); + } + + #[tokio::test(start_paused = true)] + async fn saturated_permit_obeys_the_single_operation_deadline() { + let semaphore = Arc::new(Semaphore::new(1)); + let _held = Arc::clone(&semaphore) + .acquire_owned() + .await + .expect("test semaphore open"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + let waiting = tokio::spawn(acquire_request_permit(semaphore, deadline)); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + + assert!(matches!( + waiting.await.expect("permit task"), + Err(ClientError::Timeout) + )); + } + + #[test] + fn reservation_responses_cannot_be_replayed_across_requests() { + let requested = FixedBytes32::new([0x81; 32]); + let returned = FixedBytes32::new([0x82; 32]); + let layout = SettlementLayoutDto { + taker_payment_input: 0, + provider_inputs: vec![InputPlacementDto { + quote_input_id: 0, + transaction_index: 1, + }], + quote_outputs: vec![OutputPlacementDto { + quote_output_id: 0, + transaction_index: 0, + }], + }; + let pset = SettlementPset::from_pset(&PartiallySignedTransaction::new_v2()) + .expect("valid test PSET"); + let cases = [ + ( + Request::CancelReservation { + reservation_id: requested, + }, + Response::ReservationCancelled { + status: status(returned), + }, + ), + ( + Request::BlindPset { + reservation_id: requested, + layout: layout.clone(), + pset: pset.clone(), + }, + Response::BlindedPset { + reservation_id: returned, + pset: pset.clone(), + }, + ), + ( + Request::Execute { + reservation_id: requested, + layout, + pset, + }, + Response::ExecutionAccepted { + status: status(returned), + }, + ), + ( + Request::GetReservationStatus { + reservation_id: requested, + }, + Response::ReservationStatus { + status: status(returned), + }, + ), + ]; + + for (request, response) in cases { + assert!(matches!( + validate_response_request_binding(&request, &response), + Err(ClientError::ResponseRequestMismatch) + )); + } + } + + #[test] + fn firm_quote_binding_requires_idempotency_key_and_exact_request() { + let expected_key = FixedBytes32::new([0x91; 32]); + let expected_request = quote_request(9); + assert!(!firm_quote_binding_matches( + expected_key, + &expected_request, + FixedBytes32::new([0x92; 32]), + &expected_request, + )); + + let mut changed_request = expected_request.clone(); + changed_request.maximum_input_asset_venue_fee += 1; + assert!(!firm_quote_binding_matches( + expected_key, + &expected_request, + expected_key, + &changed_request, + )); + assert!(firm_quote_binding_matches( + expected_key, + &expected_request, + expected_key, + &expected_request, + )); + } +} diff --git a/crates/deadcat-rfq-iroh/src/handler.rs b/crates/deadcat-rfq-iroh/src/handler.rs new file mode 100644 index 0000000..50a10ef --- /dev/null +++ b/crates/deadcat-rfq-iroh/src/handler.rs @@ -0,0 +1,77 @@ +//! Transport-facing request handler implemented by the RFQ service. + +use deadcat_rfq_rpc::{Request, RequestEnvelope, Response, RpcError, RpcErrorCode}; + +/// Authenticated Iroh endpoint identity of the connected client. +pub type ClientId = [u8; 32]; + +/// Dispatch target for the RFQ Iroh server adapter. +pub trait RequestHandler: Send + Sync + 'static { + /// Validate an envelope before dispatch. + /// + /// The transport always enforces schema and method-local semantic + /// validation first. Implementations may add cheap synchronous policy + /// checks; reservation authorization belongs in [`Self::handle`] and must + /// be derived from `peer` rather than wire data. + fn validate(&self, _peer: ClientId, envelope: &RequestEnvelope) -> Result<(), RpcError> { + validate_protocol_request(envelope) + } + + /// Handle one authenticated request. + /// + /// The transport cancels this future when `ServerConfig::handler_timeout` + /// elapses. An `Execute` implementation must therefore transfer any work + /// that follows its durable point of no return to daemon-owned recovery + /// before crossing that boundary; signing, persistence, and relay must not + /// depend on this future continuing to be polled. + fn handle( + &self, + peer: ClientId, + request: Request, + ) -> impl Future> + Send; +} + +pub(crate) fn validate_protocol_request(envelope: &RequestEnvelope) -> Result<(), RpcError> { + envelope.validate_version()?; + envelope.request.validate().map_err(|_| { + RpcError::new(RpcErrorCode::InvalidRequest, "invalid RFQ request") + .expect("static request error satisfies public RPC bounds") + }) +} + +#[cfg(test)] +mod tests { + use deadcat_rfq_rpc::{ + FixedBytes32, InputPlacementDto, OutputPlacementDto, RequestId, SettlementLayoutDto, + SettlementPset, + }; + use elements::pset::PartiallySignedTransaction; + + use super::*; + + #[test] + fn protocol_validation_rejects_semantically_invalid_requests() { + let envelope = RequestEnvelope::new( + RequestId(1), + Request::BlindPset { + reservation_id: FixedBytes32::new([0x55; 32]), + layout: SettlementLayoutDto { + taker_payment_input: 0, + provider_inputs: vec![InputPlacementDto { + quote_input_id: 0, + transaction_index: 0, + }], + quote_outputs: vec![OutputPlacementDto { + quote_output_id: 0, + transaction_index: 0, + }], + }, + pset: SettlementPset::from_pset(&PartiallySignedTransaction::new_v2()) + .expect("valid test PSET"), + }, + ); + + let error = validate_protocol_request(&envelope).expect_err("aliased input"); + assert_eq!(error.code(), RpcErrorCode::InvalidRequest); + } +} diff --git a/crates/deadcat-rfq-iroh/src/lib.rs b/crates/deadcat-rfq-iroh/src/lib.rs new file mode 100644 index 0000000..1db15c1 --- /dev/null +++ b/crates/deadcat-rfq-iroh/src/lib.rs @@ -0,0 +1,18 @@ +//! Native Iroh transport for authenticated Deadcat RFQ requests. +//! +//! RFQ transport uses a dedicated ALPN and exactly one request and response +//! on each bidirectional stream. Client and server identities are supplied by +//! their callers so reservation ownership remains stable across restarts. + +pub mod client; +pub mod handler; +pub mod server; + +pub use client::{Client, ClientConfig, ClientError}; +pub use handler::{ClientId, RequestHandler}; +pub use server::{DiscoveryMode, Server, ServerConfig, ServerError, SpawnedServer}; + +pub use deadcat_rfq_rpc::ALPN; + +// Keep Iroh identity types behind the transport dependency boundary. +pub use iroh::{EndpointAddr, EndpointId, SecretKey}; diff --git a/crates/deadcat-rfq-iroh/src/server.rs b/crates/deadcat-rfq-iroh/src/server.rs new file mode 100644 index 0000000..4ac332a --- /dev/null +++ b/crates/deadcat-rfq-iroh/src/server.rs @@ -0,0 +1,413 @@ +//! Bounded persistent-identity Iroh server for Deadcat RFQ RPC. + +use std::sync::Arc; +use std::time::Duration; + +use deadcat_iroh::wire::{ + self, DEFAULT_INBOUND_BUDGET_BYTES, InboundBudget, MAX_FRAME_BYTES, WireError, +}; +use deadcat_rfq_rpc::{RequestEnvelope, RpcError, RpcErrorCode, RpcOutcome, ServerEnvelope}; +use iroh::endpoint::{RecvStream, SendStream, presets}; +use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +use tokio::task::{JoinHandle, JoinSet}; + +use crate::ALPN; +use crate::handler::{ClientId, RequestHandler, validate_protocol_request}; + +#[derive(Clone, Debug)] +pub struct ServerConfig { + pub max_connections: usize, + pub max_streams_per_connection: usize, + pub max_in_flight_requests: usize, + pub inbound_budget_bytes: usize, + pub handshake_timeout: Duration, + pub request_read_timeout: Duration, + pub handler_timeout: Duration, + pub response_write_timeout: Duration, +} + +impl Default for ServerConfig { + fn default() -> Self { + Self { + max_connections: 64, + max_streams_per_connection: 16, + max_in_flight_requests: 64, + inbound_budget_bytes: DEFAULT_INBOUND_BUDGET_BYTES, + handshake_timeout: Duration::from_secs(15), + request_read_timeout: Duration::from_secs(30), + handler_timeout: Duration::from_secs(30), + response_write_timeout: Duration::from_secs(30), + } + } +} + +impl ServerConfig { + fn validate(&self) -> Result<(), ServerError> { + if self.max_connections == 0 { + return Err(ServerError::InvalidConfig( + "max_connections must be nonzero", + )); + } + if self.max_streams_per_connection == 0 { + return Err(ServerError::InvalidConfig( + "max_streams_per_connection must be nonzero", + )); + } + if self.max_in_flight_requests == 0 { + return Err(ServerError::InvalidConfig( + "max_in_flight_requests must be nonzero", + )); + } + if self.inbound_budget_bytes < MAX_FRAME_BYTES + || self.inbound_budget_bytes > usize::try_from(u32::MAX).expect("u32 fits usize") + { + return Err(ServerError::InvalidConfig( + "inbound_budget_bytes must fit at least one maximum frame and be <= u32::MAX", + )); + } + if self.max_streams_per_connection > u32::MAX as usize { + return Err(ServerError::InvalidConfig( + "max_streams_per_connection must fit u32", + )); + } + if [ + self.handshake_timeout, + self.request_read_timeout, + self.handler_timeout, + self.response_write_timeout, + ] + .contains(&Duration::ZERO) + { + return Err(ServerError::InvalidConfig("timeouts must be nonzero")); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug)] +pub enum DiscoveryMode { + N0Defaults, + Disabled, +} + +#[derive(Debug, thiserror::Error)] +pub enum ServerError { + #[error("invalid server configuration: {0}")] + InvalidConfig(&'static str), + #[error("failed to bind Iroh endpoint: {0}")] + Bind(String), + #[error("server task failed: {0}")] + Task(#[from] tokio::task::JoinError), +} + +pub struct Server { + endpoint: Endpoint, + handler: Arc, + config: ServerConfig, + inbound_budget: InboundBudget, + global_requests: Arc, +} + +impl Server { + /// Bind the RFQ endpoint using a caller-managed persistent identity key. + /// + /// The endpoint ID is also the provider identity used by quote + /// attestations, so callers must reload the same key after restart. + pub async fn bind( + secret_key: SecretKey, + discovery: DiscoveryMode, + config: ServerConfig, + handler: Arc, + ) -> Result { + config.validate()?; + let mut builder = Endpoint::builder(presets::N0) + .secret_key(secret_key) + .alpns(vec![ALPN.to_vec()]); + if matches!(discovery, DiscoveryMode::Disabled) { + builder = builder.relay_mode(RelayMode::Disabled); + } + let endpoint = builder + .bind() + .await + .map_err(|error| ServerError::Bind(error.to_string()))?; + Ok(Self { + endpoint, + handler, + inbound_budget: InboundBudget::new(config.inbound_budget_bytes), + global_requests: Arc::new(Semaphore::new(config.max_in_flight_requests)), + config, + }) + } + + #[must_use] + pub fn endpoint_id(&self) -> EndpointId { + self.endpoint.id() + } + + #[must_use] + pub fn endpoint_addr(&self) -> EndpointAddr { + self.endpoint.addr() + } + + pub async fn run(self) { + let Self { + endpoint, + handler, + config, + inbound_budget, + global_requests, + } = self; + let connection_limit = Arc::new(Semaphore::new(config.max_connections)); + let mut connections = JoinSet::new(); + + loop { + tokio::select! { + joined = connections.join_next(), if !connections.is_empty() => { + log_task_result(joined, "RFQ Iroh connection task"); + } + incoming = endpoint.accept() => { + let Some(incoming) = incoming else { + break; + }; + let permit = match Arc::clone(&connection_limit).try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + tracing::warn!(max = config.max_connections, "RFQ Iroh connection cap reached"); + incoming.refuse(); + continue; + } + }; + let handler = Arc::clone(&handler); + let config = config.clone(); + let inbound_budget = inbound_budget.clone(); + let global_requests = Arc::clone(&global_requests); + connections.spawn(async move { + let _permit = permit; + if let Err(error) = handle_connection( + incoming, + handler, + config, + inbound_budget, + global_requests, + ) + .await + { + tracing::debug!(%error, "RFQ Iroh connection ended with an error"); + } + }); + } + } + } + + while let Some(joined) = connections.join_next().await { + log_task_result(Some(joined), "RFQ Iroh connection task during shutdown"); + } + } + + #[must_use] + pub fn spawn(self) -> SpawnedServer { + let endpoint = self.endpoint.clone(); + let task = tokio::spawn(self.run()); + SpawnedServer { endpoint, task } + } +} + +pub struct SpawnedServer { + endpoint: Endpoint, + task: JoinHandle<()>, +} + +impl SpawnedServer { + pub async fn shutdown_and_join(self) -> Result<(), ServerError> { + self.endpoint.close().await; + self.task.await?; + Ok(()) + } +} + +fn log_task_result(result: Option>, label: &'static str) { + if let Some(Err(error)) = result + && !error.is_cancelled() + { + tracing::warn!(%error, task = label, "transport task panicked"); + } +} + +async fn handle_connection( + incoming: iroh::endpoint::Incoming, + handler: Arc, + config: ServerConfig, + inbound_budget: InboundBudget, + global_requests: Arc, +) -> Result<(), StreamError> { + let mut accepting = incoming + .accept() + .map_err(|error| StreamError::Transport(error.to_string()))?; + let alpn = tokio::time::timeout(config.handshake_timeout, accepting.alpn()) + .await + .map_err(|_| StreamError::Timeout("ALPN handshake"))? + .map_err(|error| StreamError::Transport(error.to_string()))?; + if alpn != ALPN { + return Err(StreamError::Transport("unexpected ALPN".into())); + } + let connection = tokio::time::timeout(config.handshake_timeout, accepting) + .await + .map_err(|_| StreamError::Timeout("connection handshake"))? + .map_err(|error| StreamError::Transport(error.to_string()))?; + connection.set_max_concurrent_uni_streams(0_u32.into()); + connection.set_max_concurrent_bi_streams( + u32::try_from(config.max_streams_per_connection) + .expect("validated above") + .into(), + ); + let peer = *connection.remote_id().as_bytes(); + let mut streams = JoinSet::new(); + + loop { + if streams.len() >= config.max_streams_per_connection { + log_task_result(streams.join_next().await, "RFQ Iroh stream task"); + continue; + } + + tokio::select! { + joined = streams.join_next(), if !streams.is_empty() => { + log_task_result(joined, "RFQ Iroh stream task"); + } + _ = connection.closed() => break, + accepted = connection.accept_bi() => { + let (send, recv) = match accepted { + Ok(streams) => streams, + Err(_) => break, + }; + let permit = match Arc::clone(&global_requests).try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + tracing::warn!("global RFQ request cap reached; refusing stream"); + drop(send); + drop(recv); + continue; + } + }; + let handler = Arc::clone(&handler); + let config = config.clone(); + let inbound_budget = inbound_budget.clone(); + streams.spawn(async move { + if let Err(error) = handle_stream( + send, + recv, + handler, + peer, + config, + inbound_budget, + permit, + ) + .await + { + tracing::debug!(%error, "RFQ Iroh stream ended with an error"); + } + }); + } + } + } + + while let Some(joined) = streams.join_next().await { + log_task_result( + Some(joined), + "RFQ Iroh stream task during connection shutdown", + ); + } + Ok(()) +} + +async fn handle_stream( + mut send: SendStream, + mut recv: RecvStream, + handler: Arc, + peer: ClientId, + config: ServerConfig, + inbound_budget: InboundBudget, + _permit: OwnedSemaphorePermit, +) -> Result<(), StreamError> { + let envelope: RequestEnvelope = tokio::time::timeout(config.request_read_timeout, async { + let envelope = wire::read_message(&mut recv, &inbound_budget).await?; + let mut trailing = [0_u8; 1]; + if recv + .read(&mut trailing) + .await + .map_err(|error| StreamError::Transport(error.to_string()))? + .is_some() + { + return Err(StreamError::TrailingData); + } + Ok::<_, StreamError>(envelope) + }) + .await + .map_err(|_| StreamError::Timeout("request read"))??; + + let request_id = envelope.request_id; + let outcome = if let Err(error) = validate_protocol_request(&envelope) { + RpcOutcome::Error { error } + } else if let Err(error) = handler.validate(peer, &envelope) { + RpcOutcome::Error { error } + } else { + match tokio::time::timeout( + config.handler_timeout, + handler.handle(peer, envelope.request), + ) + .await + { + Ok(Ok(value)) => match value.validate() { + Ok(()) => RpcOutcome::Success { value }, + Err(error) => { + tracing::error!(%error, ?peer, "RFQ handler returned an invalid response"); + RpcOutcome::Error { + error: invalid_response_rpc_error(), + } + } + }, + Ok(Err(error)) => RpcOutcome::Error { error }, + Err(_) => RpcOutcome::Error { + error: timeout_rpc_error(), + }, + } + }; + let response = ServerEnvelope::new(request_id, outcome); + tokio::time::timeout( + config.response_write_timeout, + wire::write_message(&mut send, &response), + ) + .await + .map_err(|_| StreamError::Timeout("response write"))??; + send.finish() + .map_err(|error| StreamError::Transport(error.to_string()))?; + Ok(()) +} + +fn timeout_rpc_error() -> RpcError { + RpcError::new( + RpcErrorCode::BackendUnavailable, + "request handler timed out", + ) + .expect("static timeout error satisfies public RPC bounds") +} + +fn invalid_response_rpc_error() -> RpcError { + RpcError::new( + RpcErrorCode::InternalError, + "RFQ handler returned an invalid response", + ) + .expect("static response error satisfies public RPC bounds") +} + +#[derive(Debug, thiserror::Error)] +enum StreamError { + #[error("wire error: {0}")] + Wire(#[from] WireError), + #[error("request stream contained trailing data")] + TrailingData, + #[error("{0} timed out")] + Timeout(&'static str), + #[error("transport error: {0}")] + Transport(String), +} diff --git a/crates/deadcat-rfq-iroh/tests/round_trip.rs b/crates/deadcat-rfq-iroh/tests/round_trip.rs new file mode 100644 index 0000000..7bbd53f --- /dev/null +++ b/crates/deadcat-rfq-iroh/tests/round_trip.rs @@ -0,0 +1,233 @@ +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use deadcat_rfq_iroh::{ + Client, ClientConfig, ClientError, DiscoveryMode, RequestHandler, Server, ServerConfig, +}; +use deadcat_rfq_rpc::{ + FixedBytes32, ProviderCapability, ProviderInfo, Request, RequestEnvelope, RequestId, Response, + RpcError, RpcErrorCode, +}; +use deadcat_types::LiquidNetwork; +use iroh::SecretKey; + +fn request(request_id: u64) -> RequestEnvelope { + RequestEnvelope::new(RequestId(request_id), Request::GetInfo) +} + +fn info_response(provider_endpoint: [u8; 32]) -> Response { + Response::Info { + info: ProviderInfo { + provider_endpoint: FixedBytes32::new(provider_endpoint), + network: LiquidNetwork::ElementsRegtest, + genesis_hash: "0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .expect("test block hash"), + policy_asset: "0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .expect("test asset id"), + capabilities: vec![ProviderCapability::FirmQuotes], + }, + } +} + +struct RecordingHandler { + seen_peers: Arc>>, + provider_endpoint: [u8; 32], +} + +impl RequestHandler for RecordingHandler { + async fn handle(&self, peer: [u8; 32], request: Request) -> Result { + assert_eq!(request, Request::GetInfo); + self.seen_peers.lock().expect("peer mutex").push(peer); + Ok(info_response(self.provider_endpoint)) + } +} + +async fn bind_recording_server( + seen_peers: Arc>>, +) -> ( + deadcat_rfq_iroh::SpawnedServer, + deadcat_rfq_iroh::EndpointAddr, + deadcat_rfq_iroh::EndpointId, +) { + let server_key = SecretKey::generate(); + let provider_endpoint = *server_key.public().as_bytes(); + let server = Server::bind( + server_key, + DiscoveryMode::Disabled, + ServerConfig::default(), + Arc::new(RecordingHandler { + seen_peers, + provider_endpoint, + }), + ) + .await + .expect("server bind"); + let address = server.endpoint_addr(); + let endpoint_id = server.endpoint_id(); + (server.spawn(), address, endpoint_id) +} + +#[tokio::test(flavor = "multi_thread")] +async fn unary_round_trip_passes_authenticated_peer() { + let seen_peers = Arc::new(Mutex::new(Vec::new())); + let (server, address, provider_endpoint) = bind_recording_server(Arc::clone(&seen_peers)).await; + let client_key = SecretKey::generate(); + let expected_peer = *client_key.public().as_bytes(); + let client = Client::dial_direct(address, client_key, ClientConfig::default()) + .await + .expect("client dial"); + + let response = client.call(request(11)).await.expect("RFQ response"); + assert!(matches!(response, Response::Info { .. })); + assert_eq!(client.endpoint_id().as_bytes(), &expected_peer); + assert_eq!(client.provider_endpoint_id(), provider_endpoint); + assert_eq!( + seen_peers.lock().expect("peer mutex").as_slice(), + &[expected_peer] + ); + + client.close().await; + server.shutdown_and_join().await.expect("server shutdown"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn reconnect_retains_persisted_identity_and_a_different_key_does_not() { + let seen_peers = Arc::new(Mutex::new(Vec::new())); + let (server, address, _) = bind_recording_server(Arc::clone(&seen_peers)).await; + let persistent_key = SecretKey::generate(); + let persistent_peer = *persistent_key.public().as_bytes(); + + let first = Client::dial_direct( + address.clone(), + persistent_key.clone(), + ClientConfig::default(), + ) + .await + .expect("first dial"); + first.call(request(20)).await.expect("first request"); + first.close().await; + + let second = Client::dial_direct(address.clone(), persistent_key, ClientConfig::default()) + .await + .expect("second dial"); + second.call(request(21)).await.expect("second request"); + second.close().await; + + let other_key = SecretKey::generate(); + let other_peer = *other_key.public().as_bytes(); + let other = Client::dial_direct(address, other_key, ClientConfig::default()) + .await + .expect("other dial"); + other.call(request(22)).await.expect("other request"); + other.close().await; + + assert_ne!(persistent_peer, other_peer); + assert_eq!( + seen_peers.lock().expect("peer mutex").as_slice(), + &[persistent_peer, persistent_peer, other_peer] + ); + server.shutdown_and_join().await.expect("server shutdown"); +} + +struct RejectingValidator; + +impl RequestHandler for RejectingValidator { + fn validate(&self, _peer: [u8; 32], _envelope: &RequestEnvelope) -> Result<(), RpcError> { + Err( + RpcError::new(RpcErrorCode::InvalidRequest, "rejected by validation hook") + .expect("bounded test error"), + ) + } + + async fn handle(&self, _peer: [u8; 32], _request: Request) -> Result { + panic!("validation must run before dispatch") + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn validation_error_is_returned_without_dispatch() { + let server = Server::bind( + SecretKey::generate(), + DiscoveryMode::Disabled, + ServerConfig::default(), + Arc::new(RejectingValidator), + ) + .await + .expect("server bind"); + let address = server.endpoint_addr(); + let server = server.spawn(); + let client = Client::dial_direct(address, SecretKey::generate(), ClientConfig::default()) + .await + .expect("client dial"); + + let error = client + .call(request(30)) + .await + .expect_err("validation failure"); + assert!( + matches!(error, ClientError::Rpc(error) if error.code() == RpcErrorCode::InvalidRequest) + ); + + client.close().await; + server.shutdown_and_join().await.expect("server shutdown"); +} + +struct SlowHandler; + +impl RequestHandler for SlowHandler { + async fn handle(&self, _peer: [u8; 32], _request: Request) -> Result { + tokio::time::sleep(Duration::from_secs(1)).await; + Ok(info_response([0x42; 32])) + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn handler_timeout_is_returned_as_a_typed_error() { + let config = ServerConfig { + handler_timeout: Duration::from_millis(20), + ..ServerConfig::default() + }; + let server = Server::bind( + SecretKey::generate(), + DiscoveryMode::Disabled, + config, + Arc::new(SlowHandler), + ) + .await + .expect("server bind"); + let address = server.endpoint_addr(); + let server = server.spawn(); + let client = Client::dial_direct(address, SecretKey::generate(), ClientConfig::default()) + .await + .expect("client dial"); + + let error = client.call(request(31)).await.expect_err("handler timeout"); + assert!( + matches!(error, ClientError::Rpc(error) if error.code() == RpcErrorCode::BackendUnavailable) + ); + + client.close().await; + server.shutdown_and_join().await.expect("server shutdown"); +} + +#[tokio::test] +async fn schema_version_is_rejected_before_network_dispatch() { + let seen_peers = Arc::new(Mutex::new(Vec::new())); + let (server, address, _) = bind_recording_server(Arc::clone(&seen_peers)).await; + let client = Client::dial_direct(address, SecretKey::generate(), ClientConfig::default()) + .await + .expect("client dial"); + let mut invalid = request(32); + invalid.schema_version += 1; + + let error = client.call(invalid).await.expect_err("schema failure"); + assert!( + matches!(error, ClientError::Rpc(error) if error.code() == RpcErrorCode::UnsupportedVersion) + ); + assert!(seen_peers.lock().expect("peer mutex").is_empty()); + + client.close().await; + server.shutdown_and_join().await.expect("server shutdown"); +} diff --git a/crates/deadcat-rfq-provider/src/lib.rs b/crates/deadcat-rfq-provider/src/lib.rs index 66eba3c..6e0f0d4 100644 --- a/crates/deadcat-rfq-provider/src/lib.rs +++ b/crates/deadcat-rfq-provider/src/lib.rs @@ -55,13 +55,13 @@ pub use quote::{ StaticRateRule, StaticRationalPricing, }; pub use store::{ - AuthoritativePrevout, CommitOutcome, DEFAULT_MAX_SETTLEMENT_INPUTS, - DEFAULT_MAX_SETTLEMENT_OUTPUTS, MAX_EXPIRATION_BATCH, ProviderBlindedPset, - ProviderBlindingCoordinator, ProviderBlindingError, ProviderError, ProviderSettlementValidator, - ProviderSigningCoordinator, ReservationBook, SCHEMA_VERSION, SettlementChainSource, - SettlementInputPlacement, SettlementLayout, SettlementLayoutError, SettlementLimitsError, - SettlementOutputPlacement, SettlementValidationError, SettlementValidationLimits, - SignedOutcome, SigningFinalizationError, ValidatedSigningIntent, + AuthoritativePrevout, AuthorizedReservationStatus, CommitOutcome, + DEFAULT_MAX_SETTLEMENT_INPUTS, DEFAULT_MAX_SETTLEMENT_OUTPUTS, MAX_EXPIRATION_BATCH, + ProviderBlindedPset, ProviderBlindingCoordinator, ProviderBlindingError, ProviderError, + ProviderSettlementValidator, ProviderSigningCoordinator, ReservationBook, SCHEMA_VERSION, + SettlementChainSource, SettlementInputPlacement, SettlementLayout, SettlementLayoutError, + SettlementLimitsError, SettlementOutputPlacement, SettlementValidationError, + SettlementValidationLimits, SignedOutcome, SigningFinalizationError, ValidatedSigningIntent, }; pub use wallet::{ ConfidentialDestination, DestinationPurpose, DestinationSource, InventorySnapshot, diff --git a/crates/deadcat-rfq-provider/src/store.rs b/crates/deadcat-rfq-provider/src/store.rs index d6c3cd0..70a8a43 100644 --- a/crates/deadcat-rfq-provider/src/store.rs +++ b/crates/deadcat-rfq-provider/src/store.rs @@ -775,7 +775,10 @@ impl ReservationBook { }) } - pub fn reservation( + /// Unauthenticated state-machine inspection retained only for internal + /// tests. Production callers must use [`Self::reservation_status`]. + #[cfg(test)] + pub(crate) fn reservation( &self, reservation_id: ReservationId, ) -> Result, ProviderError> { @@ -799,6 +802,52 @@ impl ReservationBook { .transpose() } + /// Return the authenticated durable status of a reservation, including + /// the exact signed bytes once signing has completed. + /// + /// The artifact is reconstructed exclusively from the provider's durable + /// record. A caller cannot use this endpoint to substitute or echo back a + /// candidate artifact, and another owner cannot observe the reservation. + pub fn reservation_status( + &self, + access: ReservationAccess, + ) -> Result { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let reservations = read.open_table(RESERVATIONS)?; + let record = reservations + .get(access.reservation_id().to_bytes().as_slice())? + .map(|value| decode_record::(value.value())) + .transpose()? + .ok_or(ProviderError::ReservationNotFound(access.reservation_id()))?; + if record.id() != access.reservation_id() { + return Err(ProviderError::CorruptState( + "reservation key and record ID disagree".to_owned(), + )); + } + record.validate()?; + if record.owner != access.owner().to_bytes() { + return Err(ProviderError::ReservationOwnerMismatch( + access.reservation_id(), + )); + } + let signed_artifact = match &record.state { + StoredReservationState::Committed { intent } => { + ensure_committed_allocations_read(&read, &record, intent.commitment)?; + None + } + StoredReservationState::Signed { intent, artifact } => { + ensure_committed_allocations_read(&read, &record, intent.commitment)?; + Some(artifact.to_domain(record.id(), SigningCommitment::new(intent.commitment))?) + } + StoredReservationState::Reserved | StoredReservationState::Released { .. } => None, + }; + Ok(AuthorizedReservationStatus { + reservation: record.to_view()?, + signed_artifact, + }) + } + /// Load the authenticated, durable inputs needed to validate a final /// settlement. The returned quote and recovery metadata are reconstructed /// from the reservation record rather than accepted from the submitter. @@ -1774,6 +1823,27 @@ pub struct SignedOutcome { recorded: bool, } +/// Owner-authenticated durable reservation state suitable for status and +/// replay responses. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AuthorizedReservationStatus { + reservation: ReservationView, + signed_artifact: Option, +} + +impl AuthorizedReservationStatus { + #[must_use] + pub const fn reservation(&self) -> &ReservationView { + &self.reservation + } + + /// Exact provider-persisted signed bytes, present only in `Signed` state. + #[must_use] + pub const fn signed_artifact(&self) -> Option<&SignedArtifact> { + self.signed_artifact.as_ref() + } +} + impl SignedOutcome { #[must_use] pub const fn artifact(&self) -> &SignedArtifact { diff --git a/crates/deadcat-rfq-provider/src/store/tests.rs b/crates/deadcat-rfq-provider/src/store/tests.rs index 8ba0a9e..fa2754d 100644 --- a/crates/deadcat-rfq-provider/src/store/tests.rs +++ b/crates/deadcat-rfq-provider/src/store/tests.rs @@ -267,6 +267,80 @@ fn reservation_is_atomic_idempotent_and_owner_authenticated() { } } +#[test] +fn reservation_status_rejects_a_different_owner() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(83); + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, inventory(124), owner(1), 1); + + assert!(matches!( + book.reservation_status(ReservationAccess::new(reservation.id(), owner(2))), + Err(ProviderError::ReservationOwnerMismatch(actual)) if actual == reservation.id() + )); +} + +#[test] +fn reserved_status_has_no_signed_artifact() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(84); + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, inventory(125), owner(1), 1); + + let status = book + .reservation_status(ReservationAccess::new( + reservation.id(), + reservation.owner(), + )) + .expect("authorized reservation status"); + assert_eq!(status.reservation(), &reservation); + assert_eq!(status.reservation().state(), ReservationState::Reserved); + assert!(status.signed_artifact().is_none()); +} + +#[test] +fn signed_status_replays_the_exact_durable_artifact() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(85); + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, inventory(126), owner(1), 1); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let committed = book + .commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ) + .expect("commit signing intent"); + let commitment = committed + .signing_job() + .expect("new signing job") + .commitment(); + let expected_bytes = vec![9, 8, 7, 6]; + let recorded = book + .record_signed( + reservation.id(), + commitment, + expected_bytes.clone(), + &UnixMillis::new(201), + ) + .expect("record signed artifact") + .artifact() + .clone(); + + let status = book + .reservation_status(access) + .expect("authorized signed status"); + assert!(matches!( + status.reservation().state(), + ReservationState::Signed { .. } + )); + let replayed = status.signed_artifact().expect("signed artifact"); + assert_eq!(replayed, &recorded); + assert_eq!(replayed.bytes(), expected_bytes); +} + #[test] fn changed_request_cannot_reuse_an_idempotency_key() { let directory = TempDir::new().expect("tempdir"); diff --git a/crates/deadcat-rfq-rpc/Cargo.toml b/crates/deadcat-rfq-rpc/Cargo.toml new file mode 100644 index 0000000..62a613b --- /dev/null +++ b/crates/deadcat-rfq-rpc/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "deadcat-rfq-rpc" +description = "Authenticated v1 wire schema and Iroh-identity attestations for Deadcat RFQ providers." +version.workspace = true +edition.workspace = true +publish.workspace = true + +[lints] +workspace = true + +[dependencies] +deadcat-types.workspace = true +elements.workspace = true +hex.workspace = true +iroh.workspace = true +postcard.workspace = true +serde.workspace = true +sha2.workspace = true +thiserror.workspace = true + +[dev-dependencies] +rand.workspace = true +serde_json.workspace = true diff --git a/crates/deadcat-rfq-rpc/src/codec.rs b/crates/deadcat-rfq-rpc/src/codec.rs new file mode 100644 index 0000000..b9d61aa --- /dev/null +++ b/crates/deadcat-rfq-rpc/src/codec.rs @@ -0,0 +1,179 @@ +use core::fmt; + +use serde::de::Error as _; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +pub(crate) mod u64_string { + use super::*; + + pub(crate) fn serialize(value: &u64, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&value.to_string()) + } + + pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + if value.is_empty() + || (value.len() > 1 && value.starts_with('0')) + || !value.bytes().all(|byte| byte.is_ascii_digit()) + { + return Err(D::Error::custom("expected canonical decimal u64 string")); + } + value.parse().map_err(D::Error::custom) + } +} + +macro_rules! fixed_hex { + ($name:ident, $size:expr) => { + #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] + pub struct $name([u8; $size]); + + impl $name { + #[must_use] + pub const fn new(bytes: [u8; $size]) -> Self { + Self(bytes) + } + + #[must_use] + pub const fn to_bytes(self) -> [u8; $size] { + self.0 + } + } + + impl fmt::Debug for $name { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(concat!(stringify!($name), "("))?; + formatter.write_str(&hex::encode(self.0))?; + formatter.write_str(")") + } + } + + impl Serialize for $name { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + if serializer.is_human_readable() { + serializer.serialize_str(&hex::encode(self.0)) + } else { + self.0.as_slice().serialize(serializer) + } + } + } + + impl<'de> Deserialize<'de> for $name { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + if deserializer.is_human_readable() { + let value = String::deserialize(deserializer)?; + if value.len() != $size * 2 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(D::Error::custom(concat!( + "expected fixed-width lowercase hex for ", + stringify!($name) + ))); + } + let decoded = hex::decode(value).map_err(D::Error::custom)?; + let bytes: [u8; $size] = decoded.try_into().map_err(|_| { + D::Error::custom(concat!("wrong byte length for ", stringify!($name))) + })?; + Ok(Self(bytes)) + } else { + let decoded = Vec::::deserialize(deserializer)?; + let bytes: [u8; $size] = decoded.try_into().map_err(|_| { + D::Error::custom(concat!("wrong byte length for ", stringify!($name))) + })?; + Ok(Self(bytes)) + } + } + } + }; +} + +fixed_hex!(FixedBytes32, 32); +fixed_hex!(FixedBytes33, 33); +fixed_hex!(FixedBytes64, 64); + +pub(crate) mod bytes_hex { + use super::*; + + pub(crate) fn serialize(value: &[u8], serializer: S) -> Result + where + S: Serializer, + { + if serializer.is_human_readable() { + serializer.serialize_str(&hex::encode(value)) + } else { + value.serialize(serializer) + } + } + + pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> + where + D: Deserializer<'de>, + { + if deserializer.is_human_readable() { + let value = String::deserialize(deserializer)?; + if value.len() % 2 != 0 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(D::Error::custom("expected canonical lowercase hex")); + } + hex::decode(value).map_err(D::Error::custom) + } else { + Vec::::deserialize(deserializer) + } + } +} + +pub(crate) mod option_bytes_hex { + use super::*; + + pub(crate) fn serialize(value: &Option>, serializer: S) -> Result + where + S: Serializer, + { + match value { + Some(value) if serializer.is_human_readable() => { + serializer.serialize_some(&hex::encode(value)) + } + Some(value) => serializer.serialize_some(value), + None => serializer.serialize_none(), + } + } + + pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result>, D::Error> + where + D: Deserializer<'de>, + { + if deserializer.is_human_readable() { + let value = Option::::deserialize(deserializer)?; + value + .map(|value| { + if value.len() % 2 != 0 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(D::Error::custom("expected canonical lowercase hex")); + } + hex::decode(value).map_err(D::Error::custom) + }) + .transpose() + } else { + Option::>::deserialize(deserializer) + } + } +} diff --git a/crates/deadcat-rfq-rpc/src/lib.rs b/crates/deadcat-rfq-rpc/src/lib.rs new file mode 100644 index 0000000..331d294 --- /dev/null +++ b/crates/deadcat-rfq-rpc/src/lib.rs @@ -0,0 +1,464 @@ +//! Strict v1 wire schema and Iroh-identity attestations for a noncustodial RFQ +//! provider. +//! +//! The transport authenticates both Iroh endpoints. This schema deliberately +//! never accepts a provider-domain `OwnerId`; the server derives it from the +//! authenticated endpoint pair with [`owner_id_from_endpoints`]. Firm quotes +//! additionally carry an application signature by the provider's stable Iroh +//! identity so they can be retained and independently verified after a stream +//! closes. + +mod codec; +mod quote; + +pub use codec::{FixedBytes32, FixedBytes33, FixedBytes64}; +pub use quote::{ + AssetAmountDto, AttestationError, BlinderRoleDto, FeePolicyDto, FeeSizeMetricDto, FirmQuoteDto, + FirmQuoteRequestDto, FirmQuoteValidationError, IdempotencyKeyDto, InputPlacementDto, + MAX_RECIPIENT_SCRIPT_BYTES, MAX_SETTLEMENT_BYTES, MAX_SETTLEMENT_INPUTS, + MAX_SETTLEMENT_OUTPUTS, OutputPlacementDto, PricingDecisionDto, PsetError, QuoteAttestation, + QuoteContextDto, QuoteExecutionDto, QuoteInputDto, QuoteKindDto, QuoteOutputDto, + QuoteOutputRoleDto, QuoteRecipientDto, RationalRateDto, ReleaseReasonDto, ReservationIdDto, + ReservationStateDto, ReservationStatusDto, SettlementLayoutDto, SettlementPset, + SignedFirmQuote, SnapshotEvidenceDto, TxOutDto, VerifiedFirmQuote, owner_id_from_endpoints, +}; + +use serde::{Deserialize, Serialize}; + +/// Iroh application protocol dedicated to RFQ traffic. +pub const ALPN: &[u8] = b"deadcat-rfq/1"; +pub const SCHEMA_VERSION: u32 = 1; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct RequestId(#[serde(with = "codec::u64_string")] pub u64); + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RequestEnvelope { + pub schema_version: u32, + pub request_id: RequestId, + pub request: Request, +} + +impl RequestEnvelope { + #[must_use] + pub const fn new(request_id: RequestId, request: Request) -> Self { + Self { + schema_version: SCHEMA_VERSION, + request_id, + request, + } + } + + pub fn validate_version(&self) -> Result<(), RpcError> { + validate_version(self.schema_version) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ServerEnvelope { + pub schema_version: u32, + pub request_id: RequestId, + pub outcome: RpcOutcome, +} + +impl ServerEnvelope { + #[must_use] + pub const fn new(request_id: RequestId, outcome: RpcOutcome) -> Self { + Self { + schema_version: SCHEMA_VERSION, + request_id, + outcome, + } + } + + pub fn validate_version(&self) -> Result<(), RpcError> { + validate_version(self.schema_version) + } +} + +fn validate_version(actual: u32) -> Result<(), RpcError> { + if actual == SCHEMA_VERSION { + Ok(()) + } else { + Err(RpcError::from_static( + RpcErrorCode::UnsupportedVersion, + "unsupported RFQ schema version", + )) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +#[allow(clippy::large_enum_variant)] +pub enum Request { + GetInfo, + RequestFirmQuote { + idempotency_key: IdempotencyKeyDto, + request: FirmQuoteRequestDto, + }, + CancelReservation { + reservation_id: ReservationIdDto, + }, + BlindPset { + reservation_id: ReservationIdDto, + layout: SettlementLayoutDto, + pset: SettlementPset, + }, + Execute { + reservation_id: ReservationIdDto, + layout: SettlementLayoutDto, + pset: SettlementPset, + }, + GetReservationStatus { + reservation_id: ReservationIdDto, + }, +} + +impl Request { + /// Perform method-local semantic checks after bounded deserialization and + /// before the runtime touches provider state or proof verification. + pub fn validate(&self) -> Result<(), FirmQuoteValidationError> { + match self { + Self::GetInfo | Self::CancelReservation { .. } | Self::GetReservationStatus { .. } => { + Ok(()) + } + Self::RequestFirmQuote { request, .. } => request.validate(), + Self::BlindPset { layout, .. } | Self::Execute { layout, .. } => layout.validate(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +#[allow(clippy::large_enum_variant)] +pub enum Response { + Info { + info: ProviderInfo, + }, + FirmQuote { + quote: SignedFirmQuote, + status: ReservationStatusDto, + }, + ReservationCancelled { + status: ReservationStatusDto, + }, + BlindedPset { + reservation_id: ReservationIdDto, + pset: SettlementPset, + }, + ExecutionAccepted { + status: ReservationStatusDto, + }, + ReservationStatus { + status: ReservationStatusDto, + }, +} + +impl Response { + /// Check all semantics that need no pinned peer, request context, clock, + /// chain source, or provider database. + pub fn validate(&self) -> Result<(), FirmQuoteValidationError> { + match self { + Self::Info { .. } | Self::BlindedPset { .. } => Ok(()), + Self::FirmQuote { quote, status } => { + quote.quote.validate_structure()?; + if quote.attestation.provider_endpoint != quote.quote.provider_endpoint { + return Err(FirmQuoteValidationError::ProviderAttestationMismatch); + } + status.validate()?; + if status.reservation_id != quote.quote.reservation_id + || status.quote_commitment != quote.quote.quote_commitment + || status.created_at_millis != quote.quote.created_at_millis + || status.accept_before_millis != quote.quote.accept_before_millis + { + return Err(FirmQuoteValidationError::QuoteStatusMismatch); + } + Ok(()) + } + Self::ReservationCancelled { status } + | Self::ExecutionAccepted { status } + | Self::ReservationStatus { status } => status.validate(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ProviderInfo { + pub provider_endpoint: FixedBytes32, + pub network: deadcat_types::LiquidNetwork, + pub genesis_hash: elements::BlockHash, + pub policy_asset: elements::AssetId, + pub capabilities: Vec, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderCapability { + FirmQuotes, + ProviderBlinding, + SettlementExecution, + DurableStatus, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +pub enum RpcOutcome { + Success { value: T }, + Error { error: RpcError }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct RpcError { + code: RpcErrorCode, + message: String, + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "optional_u64_string" + )] + retry_after_millis: Option, +} + +impl RpcError { + pub fn new( + code: RpcErrorCode, + message: impl Into, + ) -> Result { + Self::with_retry_after(code, message, None) + } + + pub fn with_retry_after( + code: RpcErrorCode, + message: impl Into, + retry_after_millis: Option, + ) -> Result { + let value = Self { + code, + message: message.into(), + retry_after_millis, + }; + value.validate()?; + Ok(value) + } + + fn from_static(code: RpcErrorCode, message: &'static str) -> Self { + debug_assert!(message.chars().count() <= MAX_ERROR_MESSAGE_CHARS); + Self { + code, + message: message.to_owned(), + retry_after_millis: None, + } + } + + pub fn validate(&self) -> Result<(), RpcErrorValidationError> { + if self.message.chars().count() > MAX_ERROR_MESSAGE_CHARS { + return Err(RpcErrorValidationError::MessageTooLong); + } + if self.retry_after_millis.is_some() + && !matches!( + self.code, + RpcErrorCode::RateLimited | RpcErrorCode::BackendUnavailable + ) + { + return Err(RpcErrorValidationError::RetryAfterNotAllowed); + } + if self.retry_after_millis == Some(0) { + return Err(RpcErrorValidationError::ZeroRetryAfter); + } + Ok(()) + } + + #[must_use] + pub const fn code(&self) -> RpcErrorCode { + self.code + } + + #[must_use] + pub fn message(&self) -> &str { + &self.message + } + + #[must_use] + pub const fn retry_after_millis(&self) -> Option { + self.retry_after_millis + } +} + +impl<'de> Deserialize<'de> for RpcError { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct Wire { + code: RpcErrorCode, + message: String, + #[serde(default, with = "optional_u64_string")] + retry_after_millis: Option, + } + + let wire = Wire::deserialize(deserializer)?; + Self::with_retry_after(wire.code, wire.message, wire.retry_after_millis) + .map_err(serde::de::Error::custom) + } +} + +mod optional_u64_string { + use serde::{Deserialize, Deserializer, Serializer}; + + pub(crate) fn serialize(value: &Option, serializer: S) -> Result + where + S: Serializer, + { + match value { + Some(value) => serializer.serialize_some(&value.to_string()), + None => serializer.serialize_none(), + } + } + + pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> + where + D: Deserializer<'de>, + { + Option::::deserialize(deserializer)? + .map(|value| { + if value.is_empty() + || (value.len() > 1 && value.starts_with('0')) + || !value.bytes().all(|byte| byte.is_ascii_digit()) + { + return Err(serde::de::Error::custom( + "expected canonical decimal u64 string", + )); + } + value.parse().map_err(serde::de::Error::custom) + }) + .transpose() + } +} + +pub const MAX_ERROR_MESSAGE_CHARS: usize = 512; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)] +pub enum RpcErrorValidationError { + #[error("public RFQ error message exceeds {MAX_ERROR_MESSAGE_CHARS} characters")] + MessageTooLong, + #[error("retry_after_millis is only valid for rate-limited or unavailable responses")] + RetryAfterNotAllowed, + #[error("retry_after_millis must be positive when present")] + ZeroRetryAfter, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RpcErrorCode { + UnsupportedVersion, + InvalidRequest, + UnsupportedMarket, + UnsupportedPair, + FillOutOfRange, + InsufficientInventory, + QuoteExpired, + IdempotencyConflict, + LiveQuoteLimit, + RateLimited, + ReservationUnavailable, + ReservationReleased, + PointOfNoReturn, + InvalidLayout, + InvalidPset, + FeePolicyRejected, + BackendUnavailable, + InternalError, +} + +#[cfg(test)] +mod tests { + use elements::pset::PartiallySignedTransaction; + + use super::*; + + #[test] + fn envelopes_are_strict_and_u64_is_a_string() { + let encoded = + serde_json::to_string(&RequestEnvelope::new(RequestId(u64::MAX), Request::GetInfo)) + .expect("encode"); + assert!(encoded.contains("\"request_id\":\"18446744073709551615\"")); + assert!(serde_json::from_str::(&encoded).is_ok()); + let extra = encoded.replacen('{', "{\"extra\":true,", 1); + assert!(serde_json::from_str::(&extra).is_err()); + assert!( + serde_json::from_str::( + r#"{"schema_version":1,"request_id":1,"request":"get_info"}"#, + ) + .is_err() + ); + } + + #[test] + fn version_is_checked() { + let mut request = RequestEnvelope::new(RequestId(1), Request::GetInfo); + request.schema_version += 1; + assert_eq!( + request.validate_version().expect_err("version").code(), + RpcErrorCode::UnsupportedVersion + ); + } + + #[test] + fn blinded_pset_response_carries_its_reservation_id() { + let reservation_id = FixedBytes32::new([0x42; 32]); + let envelope = ServerEnvelope::new( + RequestId(2), + RpcOutcome::Success { + value: Response::BlindedPset { + reservation_id, + pset: SettlementPset::from_pset(&PartiallySignedTransaction::new_v2()) + .expect("valid test PSET"), + }, + }, + ); + let encoded = serde_json::to_string(&envelope).expect("encode"); + assert!(encoded.contains(&format!( + "\"reservation_id\":\"{}\"", + hex::encode(reservation_id.to_bytes()) + ))); + assert_eq!( + serde_json::from_str::(&encoded).expect("decode"), + envelope + ); + } + + #[test] + fn public_errors_are_bounded_and_retry_hints_are_constrained() { + assert_eq!( + RpcError::new(RpcErrorCode::InvalidRequest, "x".repeat(513)), + Err(RpcErrorValidationError::MessageTooLong) + ); + assert_eq!( + RpcError::with_retry_after(RpcErrorCode::InvalidRequest, "bad", Some(10)), + Err(RpcErrorValidationError::RetryAfterNotAllowed) + ); + assert_eq!( + RpcError::with_retry_after(RpcErrorCode::RateLimited, "later", Some(0)), + Err(RpcErrorValidationError::ZeroRetryAfter) + ); + let value = RpcError::with_retry_after(RpcErrorCode::RateLimited, "later", Some(250)) + .expect("error"); + let encoded = serde_json::to_string(&value).expect("encode"); + assert!(encoded.contains("\"retry_after_millis\":\"250\"")); + assert_eq!( + serde_json::from_str::(&encoded).expect("decode"), + value + ); + let oversized = format!( + r#"{{"code":"internal_error","message":"{}"}}"#, + "é".repeat(513) + ); + assert!(serde_json::from_str::(&oversized).is_err()); + } +} diff --git a/crates/deadcat-rfq-rpc/src/quote.rs b/crates/deadcat-rfq-rpc/src/quote.rs new file mode 100644 index 0000000..da41c9e --- /dev/null +++ b/crates/deadcat-rfq-rpc/src/quote.rs @@ -0,0 +1,1836 @@ +use std::collections::BTreeSet; + +use deadcat_types::{ContractId, LiquidNetwork, serde_u64_string}; +use elements::encode::{deserialize, serialize}; +use elements::hashes::Hash as _; +use elements::pset::PartiallySignedTransaction; +use elements::secp256k1_zkp::{PublicKey, RangeProof, SurjectionProof}; +use elements::{AssetId, BlockHash, OutPoint, Script, TxOut, TxOutWitness}; +use iroh::{EndpointId, SecretKey, Signature}; +use serde::de::Error as _; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; + +use crate::codec::{FixedBytes32, FixedBytes33, FixedBytes64, bytes_hex, option_bytes_hex}; +use crate::{ALPN, SCHEMA_VERSION}; + +/// Protocol resource bounds. Runtime/domain adapters must reject any stricter +/// provider limit before doing expensive settlement work. +pub const MAX_SETTLEMENT_BYTES: usize = 1_000_000; +pub const MAX_SETTLEMENT_INPUTS: usize = 32; +pub const MAX_SETTLEMENT_OUTPUTS: usize = 32; +pub const MAX_RECIPIENT_SCRIPT_BYTES: usize = 10_000; + +const OWNER_ID_DOMAIN: &[u8] = b"deadcat/rfq/owner-id/v1"; +const QUOTE_ATTESTATION_DOMAIN: &[u8] = b"deadcat/rfq/network-firm-quote/v1"; +const SIGNED_ARTIFACT_DOMAIN: &[u8] = b"deadcat/rfq/signed-artifact/v1"; + +pub type IdempotencyKeyDto = FixedBytes32; +pub type ReservationIdDto = FixedBytes32; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct QuoteContextDto { + pub network: LiquidNetwork, + pub genesis_hash: BlockHash, + pub market: ContractId, + pub policy_asset: AssetId, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AssetAmountDto { + pub asset: AssetId, + #[serde(with = "serde_u64_string")] + pub amount: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct QuoteRecipientDto { + #[serde(with = "bytes_hex")] + pub script_pubkey: Vec, + pub blinding_public_key: FixedBytes33, +} + +impl QuoteRecipientDto { + pub fn validate(&self) -> Result<(), FirmQuoteValidationError> { + let script = Script::from(self.script_pubkey.clone()); + if script.is_empty() + || script.is_provably_unspendable() + || script.len() > MAX_RECIPIENT_SCRIPT_BYTES + { + return Err(FirmQuoteValidationError::InvalidRecipient); + } + PublicKey::from_slice(&self.blinding_public_key.to_bytes()) + .map_err(|_| FirmQuoteValidationError::InvalidBlindingKey)?; + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +pub enum QuoteKindDto { + ExactIn { + input: AssetAmountDto, + output_asset: AssetId, + #[serde(with = "serde_u64_string")] + minimum_output: u64, + }, + ExactOut { + input_asset: AssetId, + #[serde(with = "serde_u64_string")] + maximum_input: u64, + output: AssetAmountDto, + }, +} + +impl QuoteKindDto { + fn validate(self) -> Result<(), FirmQuoteValidationError> { + let (input, input_amount, output, output_amount) = match self { + Self::ExactIn { + input, + output_asset, + minimum_output, + } => (input.asset, input.amount, output_asset, minimum_output), + Self::ExactOut { + input_asset, + maximum_input, + output, + } => (input_asset, maximum_input, output.asset, output.amount), + }; + if input == output { + return Err(FirmQuoteValidationError::SameAssetPair); + } + if input_amount == 0 || output_amount == 0 { + return Err(FirmQuoteValidationError::ZeroAmount); + } + Ok(()) + } + + fn pair(self) -> (AssetId, AssetId) { + match self { + Self::ExactIn { + input, + output_asset, + .. + } => (input.asset, output_asset), + Self::ExactOut { + input_asset, + output, + .. + } => (input_asset, output.asset), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FirmQuoteRequestDto { + pub context: QuoteContextDto, + pub kind: QuoteKindDto, + pub recipient: QuoteRecipientDto, + #[serde(with = "serde_u64_string")] + pub maximum_input_asset_venue_fee: u64, +} + +impl FirmQuoteRequestDto { + pub fn validate(&self) -> Result<(), FirmQuoteValidationError> { + let market = self.context.market.creation_anchor(); + if market.is_null() || market.vout & 0xc000_0000 != 0 { + return Err(FirmQuoteValidationError::InvalidMarket); + } + self.kind.validate()?; + self.recipient.validate() + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct QuoteExecutionDto { + pub input: AssetAmountDto, + pub output: AssetAmountDto, + #[serde(with = "serde_u64_string")] + pub input_asset_venue_fee: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RationalRateDto { + #[serde(with = "serde_u64_string")] + pub numerator: u64, + #[serde(with = "serde_u64_string")] + pub denominator: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PricingDecisionDto { + pub rate: RationalRateDto, + #[serde(with = "serde_u64_string")] + pub input_asset_venue_fee: u64, + pub policy_id: FixedBytes32, + #[serde(with = "serde_u64_string")] + pub revision: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +pub enum BlinderRoleDto { + TakerPaymentInput, + ProviderInput { quote_input_id: u16 }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum QuoteOutputRoleDto { + ProviderPayment, + TakerReceive, + ProviderChange, +} + +/// Consensus `TxOut` base fields plus both confidential proof witnesses. +/// Elements consensus encoding omits the `TxOutWitness`, so carrying only the +/// base bytes would silently lose the rangeproof needed by settlement. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TxOutDto { + #[serde(with = "bytes_hex")] + pub base: Vec, + #[serde(default, with = "option_bytes_hex")] + pub surjection_proof: Option>, + #[serde(default, with = "option_bytes_hex")] + pub rangeproof: Option>, +} + +impl TxOutDto { + #[must_use] + pub fn from_txout(txout: &TxOut) -> Self { + Self { + base: serialize(txout), + surjection_proof: txout + .witness + .surjection_proof + .as_deref() + .map(SurjectionProof::serialize), + rangeproof: txout + .witness + .rangeproof + .as_deref() + .map(RangeProof::serialize), + } + } + + pub fn to_txout(&self) -> Result { + let total = self + .base + .len() + .checked_add(self.surjection_proof.as_ref().map_or(0, Vec::len)) + .and_then(|length| length.checked_add(self.rangeproof.as_ref().map_or(0, Vec::len))) + .ok_or(FirmQuoteValidationError::TxOutTooLarge)?; + if total > MAX_SETTLEMENT_BYTES { + return Err(FirmQuoteValidationError::TxOutTooLarge); + } + let mut txout = + deserialize::(&self.base).map_err(|_| FirmQuoteValidationError::InvalidTxOut)?; + txout.witness = TxOutWitness { + surjection_proof: self + .surjection_proof + .as_deref() + .map(|proof| SurjectionProof::from_slice(proof).map(Box::new)) + .transpose() + .map_err(|_| FirmQuoteValidationError::InvalidSurjectionProof)?, + rangeproof: self + .rangeproof + .as_deref() + .map(|proof| RangeProof::from_slice(proof).map(Box::new)) + .transpose() + .map_err(|_| FirmQuoteValidationError::InvalidRangeproof)?, + }; + Ok(txout) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct QuoteInputDto { + pub id: u16, + pub outpoint: OutPoint, + pub witness_utxo: TxOutDto, + pub inventory_binding: FixedBytes32, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct QuoteOutputDto { + pub id: u16, + pub role: QuoteOutputRoleDto, + pub asset: AssetId, + #[serde(with = "serde_u64_string")] + pub amount: u64, + pub destination: QuoteRecipientDto, + pub blinder: BlinderRoleDto, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SnapshotEvidenceDto { + pub block_hash: BlockHash, + pub block_height: u32, + pub snapshot_commitment: FixedBytes32, + #[serde(with = "serde_u64_string")] + pub allocation_revision: u64, + pub eligible_commitment: FixedBytes32, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FeeSizeMetricDto { + RegularVbytes, + DiscountVbytes, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FeePolicyDto { + pub policy_asset: AssetId, + #[serde(with = "serde_u64_string")] + pub minimum_sats_per_kvb: u64, + #[serde(with = "serde_u64_string")] + pub minimum_absolute_fee: u64, + #[serde(with = "serde_u64_string")] + pub maximum_transaction_weight: u64, + pub size_metric: FeeSizeMetricDto, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FirmQuoteDto { + pub reservation_id: ReservationIdDto, + /// Must equal the stable Iroh endpoint signing this quote. + pub provider_endpoint: FixedBytes32, + pub network: LiquidNetwork, + pub genesis_hash: BlockHash, + pub policy_asset: AssetId, + pub request: FirmQuoteRequestDto, + pub execution: QuoteExecutionDto, + pub pricing: PricingDecisionDto, + pub snapshot: SnapshotEvidenceDto, + pub inputs: Vec, + pub outputs: Vec, + #[serde(with = "serde_u64_string")] + pub created_at_millis: u64, + #[serde(with = "serde_u64_string")] + pub accept_before_millis: u64, + pub fee_policy: FeePolicyDto, + pub recovery_metadata_commitment: FixedBytes32, + pub quote_commitment: FixedBytes32, +} + +impl FirmQuoteDto { + pub fn validate_structure(&self) -> Result<(), FirmQuoteValidationError> { + self.request.validate()?; + if self.network != self.request.context.network + || self.genesis_hash != self.request.context.genesis_hash + || self.policy_asset != self.request.context.policy_asset + || self.fee_policy.policy_asset != self.policy_asset + { + return Err(FirmQuoteValidationError::ContextMismatch); + } + if self.created_at_millis >= self.accept_before_millis { + return Err(FirmQuoteValidationError::InvalidValidityWindow); + } + if self.fee_policy.minimum_sats_per_kvb == 0 + || self.fee_policy.maximum_transaction_weight == 0 + { + return Err(FirmQuoteValidationError::InvalidFeePolicy); + } + let pair = self.request.kind.pair(); + if self.execution.input.asset != pair.0 + || self.execution.output.asset != pair.1 + || self.execution.input.amount == 0 + || self.execution.output.amount == 0 + || self.execution.input_asset_venue_fee != self.pricing.input_asset_venue_fee + || self.execution.input_asset_venue_fee > self.request.maximum_input_asset_venue_fee + { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + match self.request.kind { + QuoteKindDto::ExactIn { + input, + minimum_output, + .. + } if self.execution.input != input || self.execution.output.amount < minimum_output => { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + QuoteKindDto::ExactOut { + maximum_input, + output, + .. + } if self.execution.output != output || self.execution.input.amount > maximum_input => { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + _ => {} + } + if self.pricing.rate.numerator == 0 + || self.pricing.rate.denominator == 0 + || gcd(self.pricing.rate.numerator, self.pricing.rate.denominator) != 1 + { + return Err(FirmQuoteValidationError::InvalidRate); + } + let fee = self.execution.input_asset_venue_fee; + let expected = match self.request.kind { + QuoteKindDto::ExactIn { .. } => { + let priced_input = self + .execution + .input + .amount + .checked_sub(fee) + .filter(|amount| *amount != 0) + .ok_or(FirmQuoteValidationError::ExecutionMismatch)?; + u128::from(priced_input) + .checked_mul(u128::from(self.pricing.rate.numerator)) + .ok_or(FirmQuoteValidationError::ExecutionMismatch)? + / u128::from(self.pricing.rate.denominator) + } + QuoteKindDto::ExactOut { .. } => { + let numerator = u128::from(self.execution.output.amount) + .checked_mul(u128::from(self.pricing.rate.denominator)) + .ok_or(FirmQuoteValidationError::ExecutionMismatch)?; + let divisor = u128::from(self.pricing.rate.numerator); + let priced_input = numerator + .checked_add(divisor - 1) + .ok_or(FirmQuoteValidationError::ExecutionMismatch)? + / divisor; + priced_input + .checked_add(u128::from(fee)) + .ok_or(FirmQuoteValidationError::ExecutionMismatch)? + } + }; + let actual = match self.request.kind { + QuoteKindDto::ExactIn { .. } => u128::from(self.execution.output.amount), + QuoteKindDto::ExactOut { .. } => u128::from(self.execution.input.amount), + }; + if expected == 0 || expected != actual { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + if self.inputs.is_empty() || self.inputs.len() > MAX_SETTLEMENT_INPUTS { + return Err(FirmQuoteValidationError::InvalidInputCount); + } + if self.outputs.len() < 2 || self.outputs.len() > MAX_SETTLEMENT_OUTPUTS { + return Err(FirmQuoteValidationError::InvalidOutputCount); + } + let mut input_ids = BTreeSet::new(); + let mut outpoints = BTreeSet::new(); + for input in &self.inputs { + if !input_ids.insert(input.id) + || !outpoints.insert(input.outpoint) + || input.outpoint.is_null() + || input.outpoint.vout & 0xc000_0000 != 0 + { + return Err(FirmQuoteValidationError::InvalidInput); + } + let prevout = input.witness_utxo.to_txout()?; + if !prevout.asset.is_confidential() + || !prevout.value.is_confidential() + || !prevout.nonce.is_confidential() + { + return Err(FirmQuoteValidationError::NonConfidentialProviderPrevout); + } + if prevout.witness.surjection_proof.is_none() { + return Err(FirmQuoteValidationError::MissingProviderSurjectionProof); + } + let rangeproof = prevout + .witness + .rangeproof + .as_deref() + .ok_or(FirmQuoteValidationError::MissingProviderRangeproof)?; + if !prevout.script_pubkey.is_v1_p2tr() { + return Err(FirmQuoteValidationError::NonP2trProviderPrevout); + } + let value_commitment = prevout + .value + .commitment() + .ok_or(FirmQuoteValidationError::NonConfidentialProviderPrevout)?; + let asset_generator = prevout + .asset + .commitment() + .ok_or(FirmQuoteValidationError::NonConfidentialProviderPrevout)?; + rangeproof + .verify( + &elements::secp256k1_zkp::Secp256k1::new(), + value_commitment, + prevout.script_pubkey.as_bytes(), + asset_generator, + ) + .map_err(|_| FirmQuoteValidationError::InvalidProviderRangeproof)?; + } + let mut output_ids = BTreeSet::new(); + let mut provider_payment = 0; + let mut taker_receive = 0; + let mut provider_change = 0; + for output in &self.outputs { + if !output_ids.insert(output.id) || output.amount == 0 { + return Err(FirmQuoteValidationError::InvalidOutput); + } + output.destination.validate()?; + match output.role { + QuoteOutputRoleDto::ProviderPayment => { + provider_payment += 1; + if output.asset != self.execution.input.asset + || output.amount != self.execution.input.amount + || output.blinder != BlinderRoleDto::TakerPaymentInput + { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + } + QuoteOutputRoleDto::TakerReceive => { + taker_receive += 1; + if output.asset != self.execution.output.asset + || output.amount != self.execution.output.amount + || output.destination != self.request.recipient + { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + if matches!(output.blinder, BlinderRoleDto::TakerPaymentInput) { + return Err(FirmQuoteValidationError::InvalidOutputBlinder); + } + } + QuoteOutputRoleDto::ProviderChange => { + provider_change += 1; + if output.asset != self.execution.output.asset { + return Err(FirmQuoteValidationError::ExecutionMismatch); + } + if matches!(output.blinder, BlinderRoleDto::TakerPaymentInput) { + return Err(FirmQuoteValidationError::InvalidOutputBlinder); + } + } + } + if let BlinderRoleDto::ProviderInput { quote_input_id } = output.blinder + && !input_ids.contains("e_input_id) + { + return Err(FirmQuoteValidationError::UnknownBlinderInput); + } + } + if provider_payment != 1 || taker_receive != 1 || provider_change > 1 { + return Err(FirmQuoteValidationError::InvalidOutputRoles); + } + Ok(()) + } +} + +const fn gcd(mut left: u64, mut right: u64) -> u64 { + while right != 0 { + let remainder = left % right; + left = right; + right = remainder; + } + left +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct InputPlacementDto { + pub quote_input_id: u16, + pub transaction_index: u16, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OutputPlacementDto { + pub quote_output_id: u16, + pub transaction_index: u16, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SettlementLayoutDto { + pub taker_payment_input: u16, + pub provider_inputs: Vec, + pub quote_outputs: Vec, +} + +impl SettlementLayoutDto { + pub fn validate(&self) -> Result<(), FirmQuoteValidationError> { + if self.provider_inputs.is_empty() + || self.provider_inputs.len() > MAX_SETTLEMENT_INPUTS + || self.quote_outputs.is_empty() + || self.quote_outputs.len() > MAX_SETTLEMENT_OUTPUTS + { + return Err(FirmQuoteValidationError::InvalidLayoutSize); + } + if usize::from(self.taker_payment_input) >= MAX_SETTLEMENT_INPUTS { + return Err(FirmQuoteValidationError::InvalidLayoutIndex); + } + let mut quote_inputs = BTreeSet::new(); + let mut transaction_inputs = BTreeSet::from([self.taker_payment_input]); + for placement in &self.provider_inputs { + if usize::from(placement.transaction_index) >= MAX_SETTLEMENT_INPUTS { + return Err(FirmQuoteValidationError::InvalidLayoutIndex); + } + if !quote_inputs.insert(placement.quote_input_id) + || !transaction_inputs.insert(placement.transaction_index) + { + return Err(FirmQuoteValidationError::AliasedLayoutInput); + } + } + let mut quote_outputs = BTreeSet::new(); + let mut transaction_outputs = BTreeSet::new(); + for placement in &self.quote_outputs { + if usize::from(placement.transaction_index) >= MAX_SETTLEMENT_OUTPUTS { + return Err(FirmQuoteValidationError::InvalidLayoutIndex); + } + if !quote_outputs.insert(placement.quote_output_id) + || !transaction_outputs.insert(placement.transaction_index) + { + return Err(FirmQuoteValidationError::AliasedLayoutOutput); + } + } + Ok(()) + } + + pub fn validate_for_quote(&self, quote: &FirmQuoteDto) -> Result<(), FirmQuoteValidationError> { + self.validate()?; + let expected_inputs = quote + .inputs + .iter() + .map(|input| input.id) + .collect::>(); + let actual_inputs = self + .provider_inputs + .iter() + .map(|input| input.quote_input_id) + .collect::>(); + let expected_outputs = quote + .outputs + .iter() + .map(|output| output.id) + .collect::>(); + let actual_outputs = self + .quote_outputs + .iter() + .map(|output| output.quote_output_id) + .collect::>(); + if expected_inputs != actual_inputs || expected_outputs != actual_outputs { + return Err(FirmQuoteValidationError::IncompleteLayout); + } + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SettlementPset(Vec); + +impl SettlementPset { + pub fn from_bytes(bytes: Vec) -> Result { + if bytes.is_empty() || bytes.len() > MAX_SETTLEMENT_BYTES { + return Err(PsetError::InvalidLength); + } + deserialize::(&bytes).map_err(|_| PsetError::InvalidPset)?; + Ok(Self(bytes)) + } + + pub fn from_pset(pset: &PartiallySignedTransaction) -> Result { + Self::from_bytes(serialize(pset)) + } + + pub fn to_pset(&self) -> Result { + deserialize(&self.0).map_err(|_| PsetError::InvalidPset) + } + + #[must_use] + pub fn as_bytes(&self) -> &[u8] { + &self.0 + } +} + +impl Serialize for SettlementPset { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + if serializer.is_human_readable() { + serializer.serialize_str(&hex::encode(&self.0)) + } else { + self.0.serialize(serializer) + } + } +} + +impl<'de> Deserialize<'de> for SettlementPset { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let bytes = if deserializer.is_human_readable() { + let value = String::deserialize(deserializer)?; + if value.len() > MAX_SETTLEMENT_BYTES * 2 + || value.len() % 2 != 0 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(D::Error::custom( + "PSET must be bounded canonical lowercase hex", + )); + } + hex::decode(value).map_err(D::Error::custom)? + } else { + Vec::::deserialize(deserializer)? + }; + Self::from_bytes(bytes).map_err(D::Error::custom) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ReleaseReasonDto { + Expired, + ClientCancelled, + ProviderRejected, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +pub enum ReservationStateDto { + Reserved, + Released { + reason: ReleaseReasonDto, + #[serde(with = "serde_u64_string")] + at_millis: u64, + }, + Committed { + signing_commitment: FixedBytes32, + #[serde(with = "serde_u64_string")] + committed_at_millis: u64, + }, + Signed { + signing_commitment: FixedBytes32, + artifact_digest: FixedBytes32, + #[serde(with = "serde_u64_string")] + committed_at_millis: u64, + #[serde(with = "serde_u64_string")] + signed_at_millis: u64, + signed_pset: SettlementPset, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ReservationStatusDto { + pub reservation_id: ReservationIdDto, + pub quote_commitment: FixedBytes32, + #[serde(with = "serde_u64_string")] + pub created_at_millis: u64, + #[serde(with = "serde_u64_string")] + pub accept_before_millis: u64, + pub state: ReservationStateDto, +} + +impl ReservationStatusDto { + pub fn validate(&self) -> Result<(), FirmQuoteValidationError> { + if self.created_at_millis >= self.accept_before_millis { + return Err(FirmQuoteValidationError::InvalidValidityWindow); + } + match &self.state { + ReservationStateDto::Reserved => {} + ReservationStateDto::Released { reason, at_millis } => { + let release_window_invalid = match reason { + ReleaseReasonDto::Expired => *at_millis < self.accept_before_millis, + ReleaseReasonDto::ClientCancelled | ReleaseReasonDto::ProviderRejected => { + *at_millis >= self.accept_before_millis + } + }; + if *at_millis < self.created_at_millis || release_window_invalid { + return Err(FirmQuoteValidationError::InvalidStatusTimeline); + } + } + ReservationStateDto::Committed { + committed_at_millis, + .. + } if *committed_at_millis < self.created_at_millis + || *committed_at_millis >= self.accept_before_millis => + { + return Err(FirmQuoteValidationError::InvalidStatusTimeline); + } + ReservationStateDto::Signed { + committed_at_millis, + signed_at_millis, + .. + } if *committed_at_millis < self.created_at_millis + || *committed_at_millis >= self.accept_before_millis + || *signed_at_millis < *committed_at_millis => + { + return Err(FirmQuoteValidationError::InvalidStatusTimeline); + } + _ => {} + } + if let ReservationStateDto::Signed { + signing_commitment, + artifact_digest, + signed_pset, + .. + } = &self.state + { + let bytes = signed_pset.as_bytes(); + let mut hasher = Sha256::new(); + hasher.update(SIGNED_ARTIFACT_DOMAIN); + hasher.update(signing_commitment.to_bytes()); + hasher.update( + u64::try_from(bytes.len()) + .map_err(|_| FirmQuoteValidationError::ArtifactDigestMismatch)? + .to_be_bytes(), + ); + hasher.update(bytes); + let expected = FixedBytes32::new(hasher.finalize().into()); + if *artifact_digest != expected { + return Err(FirmQuoteValidationError::ArtifactDigestMismatch); + } + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct QuoteAttestation { + pub provider_endpoint: FixedBytes32, + pub client_endpoint: FixedBytes32, + pub idempotency_key: IdempotencyKeyDto, + pub signature: FixedBytes64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SignedFirmQuote { + pub quote: FirmQuoteDto, + pub attestation: QuoteAttestation, +} + +#[derive(Serialize)] +struct QuoteAttestationTranscript<'a> { + schema_version: u32, + alpn: &'a [u8], + provider_endpoint: [u8; 32], + client_endpoint: [u8; 32], + idempotency_key: [u8; 32], + quote: CanonicalFirmQuoteV1, +} + +/// Frozen primitive-only normalization. Adding a field to the JSON DTO cannot +/// silently change an existing attestation: protocol evolution must explicitly +/// change this encoder and its domain/version. +#[derive(Serialize)] +struct CanonicalFirmQuoteV1 { + fields: Vec, +} + +struct CanonicalWriter(Vec); + +impl CanonicalWriter { + fn bytes(&mut self, value: &[u8]) -> Result<(), AttestationError> { + self.u64(u64::try_from(value.len()).map_err(|_| AttestationError::TranscriptEncoding)?); + self.0.extend_from_slice(value); + Ok(()) + } + + fn fixed(&mut self, value: &[u8]) { + self.0.extend_from_slice(value); + } + + fn u8(&mut self, value: u8) { + self.0.push(value); + } + + fn u16(&mut self, value: u16) { + self.0.extend_from_slice(&value.to_be_bytes()); + } + + fn u32(&mut self, value: u32) { + self.0.extend_from_slice(&value.to_be_bytes()); + } + + fn u64(&mut self, value: u64) { + self.0.extend_from_slice(&value.to_be_bytes()); + } + + fn option_bytes(&mut self, value: Option<&[u8]>) -> Result<(), AttestationError> { + match value { + Some(value) => { + self.u8(1); + self.bytes(value) + } + None => { + self.u8(0); + Ok(()) + } + } + } +} + +fn canonical_quote(quote: &FirmQuoteDto) -> Result { + fn asset(writer: &mut CanonicalWriter, value: AssetId) { + writer.fixed(&value.into_inner().to_byte_array()); + } + fn outpoint(writer: &mut CanonicalWriter, value: OutPoint) { + writer.fixed(&value.txid.to_byte_array()); + writer.u32(value.vout); + } + fn recipient( + writer: &mut CanonicalWriter, + value: &QuoteRecipientDto, + ) -> Result<(), AttestationError> { + writer.bytes(&value.script_pubkey)?; + writer.fixed(&value.blinding_public_key.to_bytes()); + Ok(()) + } + fn network(value: LiquidNetwork) -> u8 { + match value { + LiquidNetwork::Liquid => 0, + LiquidNetwork::LiquidTestnet => 1, + LiquidNetwork::ElementsRegtest => 2, + } + } + + let mut writer = CanonicalWriter(Vec::new()); + writer.fixed("e.reservation_id.to_bytes()); + writer.fixed("e.provider_endpoint.to_bytes()); + writer.u8(network(quote.network)); + writer.fixed("e.genesis_hash.to_byte_array()); + asset(&mut writer, quote.policy_asset); + writer.u8(network(quote.request.context.network)); + writer.fixed("e.request.context.genesis_hash.to_byte_array()); + outpoint(&mut writer, quote.request.context.market.creation_anchor()); + asset(&mut writer, quote.request.context.policy_asset); + match quote.request.kind { + QuoteKindDto::ExactIn { + input, + output_asset, + minimum_output, + } => { + writer.u8(0); + asset(&mut writer, input.asset); + writer.u64(input.amount); + asset(&mut writer, output_asset); + writer.u64(minimum_output); + } + QuoteKindDto::ExactOut { + input_asset, + maximum_input, + output, + } => { + writer.u8(1); + asset(&mut writer, input_asset); + writer.u64(maximum_input); + asset(&mut writer, output.asset); + writer.u64(output.amount); + } + } + recipient(&mut writer, "e.request.recipient)?; + writer.u64(quote.request.maximum_input_asset_venue_fee); + asset(&mut writer, quote.execution.input.asset); + writer.u64(quote.execution.input.amount); + asset(&mut writer, quote.execution.output.asset); + writer.u64(quote.execution.output.amount); + writer.u64(quote.execution.input_asset_venue_fee); + writer.u64(quote.pricing.rate.numerator); + writer.u64(quote.pricing.rate.denominator); + writer.u64(quote.pricing.input_asset_venue_fee); + writer.fixed("e.pricing.policy_id.to_bytes()); + writer.u64(quote.pricing.revision); + writer.fixed("e.snapshot.block_hash.to_byte_array()); + writer.u32(quote.snapshot.block_height); + writer.fixed("e.snapshot.snapshot_commitment.to_bytes()); + writer.u64(quote.snapshot.allocation_revision); + writer.fixed("e.snapshot.eligible_commitment.to_bytes()); + writer + .u64(u64::try_from(quote.inputs.len()).map_err(|_| AttestationError::TranscriptEncoding)?); + for input in "e.inputs { + writer.u16(input.id); + outpoint(&mut writer, input.outpoint); + writer.bytes(&input.witness_utxo.base)?; + writer.option_bytes(input.witness_utxo.surjection_proof.as_deref())?; + writer.option_bytes(input.witness_utxo.rangeproof.as_deref())?; + writer.fixed(&input.inventory_binding.to_bytes()); + } + writer + .u64(u64::try_from(quote.outputs.len()).map_err(|_| AttestationError::TranscriptEncoding)?); + for output in "e.outputs { + writer.u16(output.id); + writer.u8(match output.role { + QuoteOutputRoleDto::ProviderPayment => 0, + QuoteOutputRoleDto::TakerReceive => 1, + QuoteOutputRoleDto::ProviderChange => 2, + }); + asset(&mut writer, output.asset); + writer.u64(output.amount); + recipient(&mut writer, &output.destination)?; + match output.blinder { + BlinderRoleDto::TakerPaymentInput => writer.u8(0), + BlinderRoleDto::ProviderInput { quote_input_id } => { + writer.u8(1); + writer.u16(quote_input_id); + } + } + } + writer.u64(quote.created_at_millis); + writer.u64(quote.accept_before_millis); + asset(&mut writer, quote.fee_policy.policy_asset); + writer.u64(quote.fee_policy.minimum_sats_per_kvb); + writer.u64(quote.fee_policy.minimum_absolute_fee); + writer.u64(quote.fee_policy.maximum_transaction_weight); + writer.u8(match quote.fee_policy.size_metric { + FeeSizeMetricDto::RegularVbytes => 0, + FeeSizeMetricDto::DiscountVbytes => 1, + }); + writer.fixed("e.recovery_metadata_commitment.to_bytes()); + writer.fixed("e.quote_commitment.to_bytes()); + Ok(CanonicalFirmQuoteV1 { fields: writer.0 }) +} + +fn attestation_digest( + quote: &FirmQuoteDto, + provider_endpoint: FixedBytes32, + client_endpoint: FixedBytes32, + idempotency_key: IdempotencyKeyDto, +) -> Result<[u8; 32], AttestationError> { + let transcript = postcard::to_allocvec(&QuoteAttestationTranscript { + schema_version: SCHEMA_VERSION, + alpn: ALPN, + provider_endpoint: provider_endpoint.to_bytes(), + client_endpoint: client_endpoint.to_bytes(), + idempotency_key: idempotency_key.to_bytes(), + quote: canonical_quote(quote)?, + }) + .map_err(|_| AttestationError::TranscriptEncoding)?; + let mut digest = Sha256::new(); + digest.update(QUOTE_ATTESTATION_DOMAIN); + digest.update( + u64::try_from(transcript.len()) + .map_err(|_| AttestationError::TranscriptEncoding)? + .to_be_bytes(), + ); + digest.update(transcript); + Ok(digest.finalize().into()) +} + +impl SignedFirmQuote { + pub fn sign( + quote: FirmQuoteDto, + provider_key: &SecretKey, + client_endpoint: EndpointId, + idempotency_key: IdempotencyKeyDto, + ) -> Result { + quote.validate_structure()?; + let provider_endpoint = FixedBytes32::new(*provider_key.public().as_bytes()); + if quote.provider_endpoint != provider_endpoint { + return Err(AttestationError::ProviderIdentityMismatch); + } + let client_endpoint = FixedBytes32::new(*client_endpoint.as_bytes()); + let digest = + attestation_digest("e, provider_endpoint, client_endpoint, idempotency_key)?; + let signature = provider_key.sign(&digest); + Ok(Self { + quote, + attestation: QuoteAttestation { + provider_endpoint, + client_endpoint, + idempotency_key, + signature: FixedBytes64::new(signature.to_bytes()), + }, + }) + } + + fn verify_authenticity( + self, + pinned_provider: EndpointId, + authenticated_client: EndpointId, + idempotency_key: IdempotencyKeyDto, + requested: &FirmQuoteRequestDto, + ) -> Result { + self.quote.validate_structure()?; + let provider = FixedBytes32::new(*pinned_provider.as_bytes()); + let client = FixedBytes32::new(*authenticated_client.as_bytes()); + if self.quote.provider_endpoint != provider + || self.attestation.provider_endpoint != provider + || self.attestation.client_endpoint != client + || self.attestation.idempotency_key != idempotency_key + { + return Err(AttestationError::ContextMismatch); + } + if &self.quote.request != requested { + return Err(AttestationError::RequestMismatch); + } + let digest = attestation_digest(&self.quote, provider, client, idempotency_key)?; + let signature = Signature::from_bytes(&self.attestation.signature.to_bytes()); + pinned_provider + .verify(&digest, &signature) + .map_err(|_| AttestationError::InvalidSignature)?; + Ok(self) + } + + /// Verify the retained quote and additionally prove it is still inside its + /// provider-selected acceptance window at the caller's trusted wall time. + pub fn verify_at( + self, + pinned_provider: EndpointId, + authenticated_client: EndpointId, + idempotency_key: IdempotencyKeyDto, + requested: &FirmQuoteRequestDto, + now_millis: u64, + ) -> Result { + let authenticated = self.verify_authenticity( + pinned_provider, + authenticated_client, + idempotency_key, + requested, + )?; + if now_millis >= authenticated.quote.accept_before_millis { + return Err(AttestationError::Expired); + } + Ok(VerifiedFirmQuote(authenticated)) + } +} + +/// Capability produced only after provider pin, authenticated client context, +/// request equality, structural checks and signature verification all pass. +/// It is intentionally not deserializable or directly constructible. This +/// capability proves identity, request binding and structure, but not current +/// liveness; settlement callers should obtain it through `verify_at`. +#[derive(Clone, Debug)] +pub struct VerifiedFirmQuote(SignedFirmQuote); + +impl VerifiedFirmQuote { + #[must_use] + pub const fn signed(&self) -> &SignedFirmQuote { + &self.0 + } + + #[must_use] + pub const fn quote(&self) -> &FirmQuoteDto { + &self.0.quote + } +} + +#[must_use] +pub fn owner_id_from_endpoints( + provider_endpoint: EndpointId, + client_endpoint: EndpointId, +) -> FixedBytes32 { + let mut digest = Sha256::new(); + digest.update(OWNER_ID_DOMAIN); + digest.update(provider_endpoint.as_bytes()); + digest.update(client_endpoint.as_bytes()); + FixedBytes32::new(digest.finalize().into()) +} + +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +pub enum PsetError { + #[error("settlement PSET must be nonempty and no larger than {MAX_SETTLEMENT_BYTES} bytes")] + InvalidLength, + #[error("invalid PSET encoding")] + InvalidPset, +} + +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +pub enum FirmQuoteValidationError { + #[error("invalid market outpoint")] + InvalidMarket, + #[error("trade assets must be distinct")] + SameAssetPair, + #[error("trade amounts must be nonzero")] + ZeroAmount, + #[error("invalid confidential recipient")] + InvalidRecipient, + #[error("invalid recipient blinding public key")] + InvalidBlindingKey, + #[error("quote chain or policy context mismatch")] + ContextMismatch, + #[error("invalid quote validity window")] + InvalidValidityWindow, + #[error("invalid fee policy")] + InvalidFeePolicy, + #[error("execution does not satisfy the request or contribution")] + ExecutionMismatch, + #[error("invalid normalized rational rate")] + InvalidRate, + #[error("invalid provider input count")] + InvalidInputCount, + #[error("invalid quote output count")] + InvalidOutputCount, + #[error("invalid or duplicate provider input")] + InvalidInput, + #[error("invalid or duplicate quote output")] + InvalidOutput, + #[error("invalid quote output roles")] + InvalidOutputRoles, + #[error("quote output refers to an unknown provider blinder input")] + UnknownBlinderInput, + #[error("provider-funded output must use a provider input blinder")] + InvalidOutputBlinder, + #[error("serialized TxOut exceeds the settlement bound")] + TxOutTooLarge, + #[error("invalid TxOut base encoding")] + InvalidTxOut, + #[error("invalid surjection proof")] + InvalidSurjectionProof, + #[error("invalid rangeproof")] + InvalidRangeproof, + #[error("provider prevout asset, value, and nonce must all be confidential")] + NonConfidentialProviderPrevout, + #[error("provider prevout is missing its surjection proof")] + MissingProviderSurjectionProof, + #[error("provider prevout is missing its rangeproof")] + MissingProviderRangeproof, + #[error("provider prevout must use a v1 P2TR script")] + NonP2trProviderPrevout, + #[error("provider prevout rangeproof does not verify against its commitments and script")] + InvalidProviderRangeproof, + #[error("invalid settlement layout size")] + InvalidLayoutSize, + #[error("settlement placement index exceeds the v1 resource limit")] + InvalidLayoutIndex, + #[error("settlement input placement aliases or repeats an input")] + AliasedLayoutInput, + #[error("settlement output placement aliases or repeats an output")] + AliasedLayoutOutput, + #[error("settlement layout does not cover every quote object exactly once")] + IncompleteLayout, + #[error("signed settlement bytes do not match their durable artifact digest")] + ArtifactDigestMismatch, + #[error("firm quote provider does not match its attestation signer")] + ProviderAttestationMismatch, + #[error("firm quote and reservation status describe different durable records")] + QuoteStatusMismatch, + #[error("reservation status timestamps are not monotonic or violate the acceptance window")] + InvalidStatusTimeline, +} + +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +pub enum AttestationError { + #[error("firm quote is structurally invalid: {0}")] + InvalidQuote(#[from] FirmQuoteValidationError), + #[error("provider endpoint does not match the signing identity")] + ProviderIdentityMismatch, + #[error("attestation endpoint or idempotency context mismatch")] + ContextMismatch, + #[error("attested quote does not match the requested terms")] + RequestMismatch, + #[error("could not encode canonical quote attestation transcript")] + TranscriptEncoding, + #[error("invalid provider quote signature")] + InvalidSignature, + #[error("firm quote acceptance window has elapsed")] + Expired, +} + +#[cfg(test)] +mod tests { + use deadcat_types::ContractId; + use elements::confidential::{Asset, AssetBlindingFactor, Nonce, Value, ValueBlindingFactor}; + use elements::hashes::Hash as _; + use elements::secp256k1_zkp::{Keypair, Secp256k1, SecretKey as SecpSecretKey}; + use elements::{RangeProofMessage, TxOut, TxOutSecrets, Txid}; + use rand::SeedableRng as _; + use rand::rngs::StdRng; + + use super::*; + use crate::{Request, Response}; + + fn asset(marker: u8) -> AssetId { + AssetId::from_slice(&[marker; 32]).expect("asset") + } + + fn outpoint(marker: u8, vout: u32) -> OutPoint { + OutPoint::new(Txid::from_byte_array([marker; 32]), vout) + } + + fn recipient(marker: u8) -> QuoteRecipientDto { + let secret = SecpSecretKey::from_slice(&[marker; 32]).expect("secret"); + let public = PublicKey::from_secret_key(&Secp256k1::new(), &secret); + QuoteRecipientDto { + script_pubkey: vec![0x51, marker], + blinding_public_key: FixedBytes33::new(public.serialize()), + } + } + + fn confidential_p2tr_txout(asset: AssetId, amount: u64) -> TxOut { + let secp = Secp256k1::new(); + let spend_secret = SecpSecretKey::from_slice(&[101; 32]).expect("spend secret"); + let spend_keypair = Keypair::from_secret_key(&secp, &spend_secret); + let (internal_key, _) = spend_keypair.x_only_public_key(); + let blinding_secret = SecpSecretKey::from_slice(&[102; 32]).expect("blinding secret"); + let blinding_public_key = PublicKey::from_secret_key(&secp, &blinding_secret); + let explicit = TxOut { + asset: Asset::Explicit(asset), + value: Value::Explicit(amount), + nonce: Nonce::Null, + script_pubkey: Script::new_v1_p2tr(&secp, internal_key, None), + witness: TxOutWitness::empty(), + }; + let mut rng = StdRng::from_seed([103; 32]); + explicit + .to_non_last_confidential( + &mut rng, + &secp, + blinding_public_key, + &[TxOutSecrets::new( + asset, + AssetBlindingFactor::zero(), + amount, + ValueBlindingFactor::zero(), + )], + ) + .expect("confidential provider prevout") + .0 + } + + fn quote(provider: EndpointId) -> FirmQuoteDto { + let input_asset = asset(1); + let output_asset = asset(2); + let taker = recipient(3); + let provider_recipient = recipient(4); + let genesis = BlockHash::from_byte_array([9; 32]); + let request = FirmQuoteRequestDto { + context: QuoteContextDto { + network: LiquidNetwork::ElementsRegtest, + genesis_hash: genesis, + market: ContractId::new(outpoint(5, 0)), + policy_asset: input_asset, + }, + kind: QuoteKindDto::ExactIn { + input: AssetAmountDto { + asset: input_asset, + amount: 100, + }, + output_asset, + minimum_output: 170, + }, + recipient: taker.clone(), + maximum_input_asset_venue_fee: 10, + }; + FirmQuoteDto { + reservation_id: FixedBytes32::new([6; 32]), + provider_endpoint: FixedBytes32::new(*provider.as_bytes()), + network: LiquidNetwork::ElementsRegtest, + genesis_hash: genesis, + policy_asset: input_asset, + request, + execution: QuoteExecutionDto { + input: AssetAmountDto { + asset: input_asset, + amount: 100, + }, + output: AssetAmountDto { + asset: output_asset, + amount: 180, + }, + input_asset_venue_fee: 10, + }, + pricing: PricingDecisionDto { + rate: RationalRateDto { + numerator: 2, + denominator: 1, + }, + input_asset_venue_fee: 10, + policy_id: FixedBytes32::new([7; 32]), + revision: 1, + }, + snapshot: SnapshotEvidenceDto { + block_hash: BlockHash::from_byte_array([8; 32]), + block_height: 10, + snapshot_commitment: FixedBytes32::new([9; 32]), + allocation_revision: 2, + eligible_commitment: FixedBytes32::new([10; 32]), + }, + inputs: vec![QuoteInputDto { + id: 1, + outpoint: outpoint(11, 0), + witness_utxo: TxOutDto::from_txout(&confidential_p2tr_txout(output_asset, 200)), + inventory_binding: FixedBytes32::new([12; 32]), + }], + outputs: vec![ + QuoteOutputDto { + id: 1, + role: QuoteOutputRoleDto::ProviderPayment, + asset: input_asset, + amount: 100, + destination: provider_recipient.clone(), + blinder: BlinderRoleDto::TakerPaymentInput, + }, + QuoteOutputDto { + id: 2, + role: QuoteOutputRoleDto::TakerReceive, + asset: output_asset, + amount: 180, + destination: taker, + blinder: BlinderRoleDto::ProviderInput { quote_input_id: 1 }, + }, + QuoteOutputDto { + id: 3, + role: QuoteOutputRoleDto::ProviderChange, + asset: output_asset, + amount: 20, + destination: provider_recipient, + blinder: BlinderRoleDto::ProviderInput { quote_input_id: 1 }, + }, + ], + created_at_millis: 1_000, + accept_before_millis: 31_000, + fee_policy: FeePolicyDto { + policy_asset: input_asset, + minimum_sats_per_kvb: 100, + minimum_absolute_fee: 10, + maximum_transaction_weight: 100_000, + size_metric: FeeSizeMetricDto::DiscountVbytes, + }, + recovery_metadata_commitment: FixedBytes32::new([13; 32]), + quote_commitment: FixedBytes32::new([14; 32]), + } + } + + #[test] + fn fixed_key_attestation_rejects_tampering_and_wrong_context() { + let provider_key = SecretKey::from_bytes(&[21; 32]); + let client_key = SecretKey::from_bytes(&[22; 32]); + let wrong_client = SecretKey::from_bytes(&[23; 32]); + let idempotency = FixedBytes32::new([24; 32]); + let request = quote(provider_key.public()).request; + let signed = SignedFirmQuote::sign( + quote(provider_key.public()), + &provider_key, + client_key.public(), + idempotency, + ) + .expect("sign"); + assert_eq!( + hex::encode(signed.attestation.signature.to_bytes()), + "9f7255b6022ba758f6fd90038b4d0f7fadff6aa3de8dd26decc990650f51883ba55033bf383ca08d97115290fe3aef2fcc6a6de09f760534458a38ce8b00750f" + ); + assert!( + signed + .clone() + .verify_at( + provider_key.public(), + client_key.public(), + idempotency, + &request, + 30_999, + ) + .is_ok() + ); + assert_eq!( + signed + .clone() + .verify_at( + provider_key.public(), + wrong_client.public(), + idempotency, + &request, + 2_000, + ) + .expect_err("wrong client"), + AttestationError::ContextMismatch + ); + assert_eq!( + signed + .clone() + .verify_at( + SecretKey::from_bytes(&[25; 32]).public(), + client_key.public(), + idempotency, + &request, + 2_000, + ) + .expect_err("wrong provider"), + AttestationError::ContextMismatch + ); + assert_eq!( + signed + .clone() + .verify_at( + provider_key.public(), + client_key.public(), + FixedBytes32::new([26; 32]), + &request, + 2_000, + ) + .expect_err("wrong idempotency key"), + AttestationError::ContextMismatch + ); + let mut changed_request = request.clone(); + changed_request.maximum_input_asset_venue_fee += 1; + assert_eq!( + signed + .clone() + .verify_at( + provider_key.public(), + client_key.public(), + idempotency, + &changed_request, + 2_000, + ) + .expect_err("changed request"), + AttestationError::RequestMismatch + ); + assert_eq!( + signed + .clone() + .verify_at( + provider_key.public(), + client_key.public(), + idempotency, + &request, + 31_000, + ) + .expect_err("expired"), + AttestationError::Expired + ); + let mut tampered = signed; + tampered.quote.accept_before_millis += 1; + assert_eq!( + tampered + .verify_at( + provider_key.public(), + client_key.public(), + idempotency, + &request, + 2_000, + ) + .expect_err("tampered"), + AttestationError::InvalidSignature + ); + } + + #[test] + fn owner_ids_are_scoped_to_both_endpoints() { + let provider = SecretKey::from_bytes(&[31; 32]).public(); + let other_provider = SecretKey::from_bytes(&[32; 32]).public(); + let client = SecretKey::from_bytes(&[33; 32]).public(); + assert_ne!( + owner_id_from_endpoints(provider, client), + owner_id_from_endpoints(other_provider, client) + ); + } + + #[test] + fn dto_json_is_strict_and_uses_canonical_strings() { + let provider = SecretKey::from_bytes(&[41; 32]).public(); + let encoded = serde_json::to_string("e(provider).request).expect("encode"); + assert!(encoded.contains("\"amount\":\"100\"")); + assert!(serde_json::from_str::(&encoded).is_ok()); + let extra = encoded.replacen('{', "{\"extra\":0,", 1); + assert!(serde_json::from_str::(&extra).is_err()); + let numeric = encoded.replacen("\"100\"", "100", 1); + assert!(serde_json::from_str::(&numeric).is_err()); + let fixed = serde_json::to_string(&FixedBytes32::new([0xab; 32])).expect("fixed"); + assert!(serde_json::from_str::(&fixed.to_uppercase()).is_err()); + } + + #[test] + fn txout_roundtrip_preserves_rangeproof_witness() { + let asset_id = asset(51); + let script = Script::from(vec![0x51]); + let secp = Secp256k1::new(); + let abf = AssetBlindingFactor::from_slice(&[1; 32]).expect("abf"); + let vbf = ValueBlindingFactor::from_slice(&[2; 32]).expect("vbf"); + let rewind = SecpSecretKey::from_slice(&[3; 32]).expect("rewind"); + let message = RangeProofMessage { + asset: asset_id, + bf: abf, + }; + let (value, proof) = Value::Explicit(42) + .blind_with_shared_secret(&secp, vbf, rewind, &script, &message) + .expect("rangeproof"); + let txout = TxOut { + asset: Asset::Explicit(asset_id), + value, + nonce: Nonce::Null, + script_pubkey: script, + witness: TxOutWitness { + surjection_proof: None, + rangeproof: Some(Box::new(proof)), + }, + }; + let dto = TxOutDto::from_txout(&txout); + let json = serde_json::to_string(&dto).expect("encode"); + let decoded: TxOutDto = serde_json::from_str(&json).expect("decode"); + assert_eq!(decoded.to_txout().expect("txout"), txout); + } + + #[test] + fn pset_hex_is_canonical_bounded_and_checked() { + let pset = + SettlementPset::from_pset(&PartiallySignedTransaction::new_v2()).expect("valid PSET"); + let json = serde_json::to_string(&pset).expect("encode"); + assert!(serde_json::from_str::(&json).is_ok()); + assert!(serde_json::from_str::(&json.to_uppercase()).is_err()); + let oversized = format!("\"{}\"", "00".repeat(MAX_SETTLEMENT_BYTES + 1)); + assert!(serde_json::from_str::(&oversized).is_err()); + assert_eq!( + SettlementPset::from_bytes(vec![1, 2, 3]), + Err(PsetError::InvalidPset) + ); + } + + #[test] + fn layout_is_injective_and_complete_for_quote() { + let provider = SecretKey::from_bytes(&[61; 32]).public(); + let quote = quote(provider); + let layout = SettlementLayoutDto { + taker_payment_input: 0, + provider_inputs: vec![InputPlacementDto { + quote_input_id: 1, + transaction_index: 1, + }], + quote_outputs: vec![ + OutputPlacementDto { + quote_output_id: 1, + transaction_index: 0, + }, + OutputPlacementDto { + quote_output_id: 2, + transaction_index: 1, + }, + OutputPlacementDto { + quote_output_id: 3, + transaction_index: 2, + }, + ], + }; + assert!(layout.validate_for_quote("e).is_ok()); + let mut aliased = layout.clone(); + aliased.provider_inputs[0].transaction_index = 0; + assert_eq!( + aliased.validate().expect_err("alias"), + FirmQuoteValidationError::AliasedLayoutInput + ); + let mut incomplete = layout; + incomplete.quote_outputs.pop(); + assert_eq!( + incomplete + .validate_for_quote("e) + .expect_err("incomplete"), + FirmQuoteValidationError::IncompleteLayout + ); + } + + #[test] + fn structure_checks_rate_and_blinder_semantics() { + let provider = SecretKey::from_bytes(&[71; 32]).public(); + let mut value = quote(provider); + assert!(value.validate_structure().is_ok()); + value.execution.output.amount += 1; + assert_eq!( + value.validate_structure().expect_err("rounding"), + FirmQuoteValidationError::ExecutionMismatch + ); + let mut value = quote(provider); + value.outputs[1].blinder = BlinderRoleDto::TakerPaymentInput; + assert_eq!( + value.validate_structure().expect_err("blinder"), + FirmQuoteValidationError::InvalidOutputBlinder + ); + } + + #[test] + fn provider_prevouts_require_the_confidential_p2tr_proof_profile() { + let provider = SecretKey::from_bytes(&[72; 32]).public(); + let valid = quote(provider); + assert!(valid.validate_structure().is_ok()); + + let mut explicit = valid.clone(); + let mut prevout = explicit.inputs[0] + .witness_utxo + .to_txout() + .expect("fixture prevout"); + prevout.asset = Asset::Explicit(valid.execution.output.asset); + explicit.inputs[0].witness_utxo = TxOutDto::from_txout(&prevout); + assert_eq!( + explicit.validate_structure().expect_err("explicit asset"), + FirmQuoteValidationError::NonConfidentialProviderPrevout + ); + + let mut missing_surjection = valid.clone(); + missing_surjection.inputs[0].witness_utxo.surjection_proof = None; + assert_eq!( + missing_surjection + .validate_structure() + .expect_err("missing surjection proof"), + FirmQuoteValidationError::MissingProviderSurjectionProof + ); + + let mut missing_rangeproof = valid.clone(); + missing_rangeproof.inputs[0].witness_utxo.rangeproof = None; + assert_eq!( + missing_rangeproof + .validate_structure() + .expect_err("missing rangeproof"), + FirmQuoteValidationError::MissingProviderRangeproof + ); + + let mut non_p2tr = valid.clone(); + let mut prevout = non_p2tr.inputs[0] + .witness_utxo + .to_txout() + .expect("fixture prevout"); + prevout.script_pubkey = Script::from(vec![0x51]); + non_p2tr.inputs[0].witness_utxo = TxOutDto::from_txout(&prevout); + assert_eq!( + non_p2tr.validate_structure().expect_err("non-P2TR"), + FirmQuoteValidationError::NonP2trProviderPrevout + ); + + let mut wrong_p2tr = valid; + let mut prevout = wrong_p2tr.inputs[0] + .witness_utxo + .to_txout() + .expect("fixture prevout"); + let secp = Secp256k1::new(); + let replacement_secret = SecpSecretKey::from_slice(&[104; 32]).expect("replacement key"); + let replacement_pair = Keypair::from_secret_key(&secp, &replacement_secret); + prevout.script_pubkey = + Script::new_v1_p2tr(&secp, replacement_pair.x_only_public_key().0, None); + wrong_p2tr.inputs[0].witness_utxo = TxOutDto::from_txout(&prevout); + assert_eq!( + wrong_p2tr + .validate_structure() + .expect_err("rangeproof script binding"), + FirmQuoteValidationError::InvalidProviderRangeproof + ); + } + + #[test] + fn signed_status_recomputes_the_provider_artifact_digest() { + let pset = + SettlementPset::from_pset(&PartiallySignedTransaction::new_v2()).expect("valid PSET"); + let signing_commitment = FixedBytes32::new([81; 32]); + let mut hasher = Sha256::new(); + hasher.update(SIGNED_ARTIFACT_DOMAIN); + hasher.update(signing_commitment.to_bytes()); + hasher.update((pset.as_bytes().len() as u64).to_be_bytes()); + hasher.update(pset.as_bytes()); + let artifact_digest = FixedBytes32::new(hasher.finalize().into()); + let status = ReservationStatusDto { + reservation_id: FixedBytes32::new([82; 32]), + quote_commitment: FixedBytes32::new([83; 32]), + created_at_millis: 1_000, + accept_before_millis: 2_000, + state: ReservationStateDto::Signed { + signing_commitment, + artifact_digest, + committed_at_millis: 1_500, + signed_at_millis: 1_600, + signed_pset: pset, + }, + }; + assert!(status.validate().is_ok()); + let mut tampered = status.clone(); + let ReservationStateDto::Signed { + artifact_digest, .. + } = &mut tampered.state + else { + unreachable!("signed fixture") + }; + *artifact_digest = FixedBytes32::new([84; 32]); + assert_eq!( + tampered.validate().expect_err("digest mismatch"), + FirmQuoteValidationError::ArtifactDigestMismatch + ); + let mut invalid_time = status; + let ReservationStateDto::Signed { + signed_at_millis, .. + } = &mut invalid_time.state + else { + unreachable!("signed fixture") + }; + *signed_at_millis = 1_499; + assert_eq!( + invalid_time.validate().expect_err("time regression"), + FirmQuoteValidationError::InvalidStatusTimeline + ); + } + + #[test] + fn released_status_enforces_reason_specific_acceptance_window() { + let status = |reason, at_millis| ReservationStatusDto { + reservation_id: FixedBytes32::new([85; 32]), + quote_commitment: FixedBytes32::new([86; 32]), + created_at_millis: 1_000, + accept_before_millis: 2_000, + state: ReservationStateDto::Released { reason, at_millis }, + }; + + assert!(status(ReleaseReasonDto::Expired, 2_000).validate().is_ok()); + assert_eq!( + status(ReleaseReasonDto::Expired, 1_999) + .validate() + .expect_err("early expiry"), + FirmQuoteValidationError::InvalidStatusTimeline + ); + assert!( + status(ReleaseReasonDto::ClientCancelled, 1_999) + .validate() + .is_ok() + ); + assert_eq!( + status(ReleaseReasonDto::ClientCancelled, 2_000) + .validate() + .expect_err("late cancellation"), + FirmQuoteValidationError::InvalidStatusTimeline + ); + assert_eq!( + status(ReleaseReasonDto::ProviderRejected, 2_000) + .validate() + .expect_err("late provider rejection"), + FirmQuoteValidationError::InvalidStatusTimeline + ); + assert_eq!( + status(ReleaseReasonDto::ProviderRejected, 999) + .validate() + .expect_err("release before creation"), + FirmQuoteValidationError::InvalidStatusTimeline + ); + } + + #[test] + fn top_level_semantic_hooks_reject_inconsistent_payloads() { + let provider_key = SecretKey::from_bytes(&[91; 32]); + let client = SecretKey::from_bytes(&[92; 32]).public(); + let idempotency = FixedBytes32::new([93; 32]); + let quote = quote(provider_key.public()); + assert!( + Request::RequestFirmQuote { + idempotency_key: idempotency, + request: quote.request.clone(), + } + .validate() + .is_ok() + ); + let signed = + SignedFirmQuote::sign(quote.clone(), &provider_key, client, idempotency).expect("sign"); + let status = ReservationStatusDto { + reservation_id: quote.reservation_id, + quote_commitment: quote.quote_commitment, + created_at_millis: quote.created_at_millis, + accept_before_millis: quote.accept_before_millis, + state: ReservationStateDto::Reserved, + }; + let response = Response::FirmQuote { + quote: signed, + status, + }; + assert!(response.validate().is_ok()); + let Response::FirmQuote { mut quote, status } = response else { + unreachable!("firm quote fixture") + }; + quote.attestation.provider_endpoint = FixedBytes32::new([94; 32]); + assert_eq!( + Response::FirmQuote { quote, status } + .validate() + .expect_err("attestation provider mismatch"), + FirmQuoteValidationError::ProviderAttestationMismatch + ); + } +} diff --git a/justfile b/justfile index 44d78ab..ddd326e 100644 --- a/justfile +++ b/justfile @@ -73,7 +73,9 @@ regtest-process-boundary: generate regtest: regtest-market-ab regtest-multi-market regtest-backend-equivalence regtest-rfq-settlement regtest-rfq-wallet-source regtest-process-boundary wasm-check: - NIX_HARDENING_ENABLE=pic cargo check --locked -p deadcat-iroh --lib --target wasm32-unknown-unknown + NIX_HARDENING_ENABLE=pic cargo check --locked \ + -p deadcat-iroh -p deadcat-rfq-rpc -p deadcat-rfq-iroh \ + --lib --target wasm32-unknown-unknown ci-checks: fmt-check clippy test wasm-check