diff --git a/core/archipelago/src/api/handler/mod.rs b/core/archipelago/src/api/handler/mod.rs index f615912d..77d7b8c1 100644 --- a/core/archipelago/src/api/handler/mod.rs +++ b/core/archipelago/src/api/handler/mod.rs @@ -1,16 +1,16 @@ -mod purchase; -mod cloud_purchase; mod blob; mod cdp; +mod cloud_purchase; mod content; -mod registered_media; -mod rental_playback; mod dwn; mod model_proxy; mod node_message; mod proxy; +mod purchase; +mod registered_media; mod remote_input; mod remote_relay; +mod rental_playback; mod routstr_proxy; mod websocket; @@ -389,7 +389,9 @@ impl ApiHandler { let method = req.method().clone(); if path.starts_with("/api/rental-playback/") { - return self.handle_local_rental_request(&method, &path, req.headers()).await; + return self + .handle_local_rental_request(&method, &path, req.headers()) + .await; } // Handle CORS preflight for all routes @@ -453,13 +455,28 @@ impl ApiHandler { } // Purchase routes bound the original body before the generic buffer. - if method == Method::POST && matches!(path.as_str(), - crate::content_purchase_protocol::OFFER_ROUTE | crate::content_purchase_protocol::ACCEPT_ROUTE - | crate::content_purchase_protocol::SETTLE_ROUTE | crate::content_purchase_protocol::STATUS_ROUTE - | crate::content_purchase_protocol::CANCEL_ROUTE) { + if method == Method::POST + && matches!( + path.as_str(), + crate::content_purchase_protocol::PREPARE_OFFER_ROUTE + | crate::content_purchase_protocol::OFFER_ROUTE + | crate::content_purchase_protocol::ACCEPT_ROUTE + | crate::content_purchase_protocol::SETTLE_ROUTE + | crate::content_purchase_protocol::STATUS_ROUTE + | crate::content_purchase_protocol::CANCEL_ROUTE + ) + { return self.handle_purchase_request(req).await; } + if method == Method::POST + && path.starts_with("/content/registered_") + && path.contains("/rental/") + && (path.ends_with("/prepare") || path.ends_with("/start")) + { + return self.handle_rental_control(req).await; + } + // Convert body to bytes for non-WS routes let headers = req.headers().clone(); let query_string = req.uri().query().map(|s| s.to_string()).unwrap_or_default(); @@ -649,21 +666,31 @@ impl ApiHandler { // falls outside `connect-src`). Session-authenticated so only // the logged-in node owner can spin up fetches. (Method::GET, "/api/node-app-catalog") => { - if !self.is_authenticated(&headers).await { return Ok(Self::unauthorized()); } + if !self.is_authenticated(&headers).await { + return Ok(Self::unauthorized()); + } let data_dir = self.config.data_dir.clone(); let result = tokio::task::spawn_blocking(move || { crate::container::node_catalog::verified_body(&data_dir) - }).await.unwrap_or_else(|error| Err(anyhow::anyhow!(error))); + }) + .await + .unwrap_or_else(|error| Err(anyhow::anyhow!(error))); let (status, body) = match result { Ok(Some(body)) => (StatusCode::OK, body), Ok(None) => (StatusCode::NOT_FOUND, "{}".to_owned()), Err(error) => { tracing::warn!("Node demo catalog rejected: {error}"); - (StatusCode::CONFLICT, "{\"error\":\"Node demo catalog is unavailable\"}".to_owned()) - }, + ( + StatusCode::CONFLICT, + "{\"error\":\"Node demo catalog is unavailable\"}".to_owned(), + ) + } }; - Ok(Response::builder().status(status).header("Content-Type", "application/json") - .header("Cache-Control", "private, no-store").body(hyper::Body::from(body))?) + Ok(Response::builder() + .status(status) + .header("Content-Type", "application/json") + .header("Cache-Control", "private, no-store") + .body(hyper::Body::from(body))?) } (Method::GET, "/api/app-catalog") => { diff --git a/core/archipelago/src/api/handler/purchase.rs b/core/archipelago/src/api/handler/purchase.rs index 1a469584..74697a10 100644 --- a/core/archipelago/src/api/handler/purchase.rs +++ b/core/archipelago/src/api/handler/purchase.rs @@ -23,7 +23,8 @@ impl ApiHandler { request.method() == Method::POST && matches!( path.as_str(), - protocol::OFFER_ROUTE + protocol::PREPARE_OFFER_ROUTE + | protocol::OFFER_ROUTE | protocol::ACCEPT_ROUTE | protocol::SETTLE_ROUTE | protocol::STATUS_ROUTE @@ -59,6 +60,32 @@ impl ApiHandler { )?; let data_dir = &self.config.data_dir; let result = match path.as_str() { + protocol::PREPARE_OFFER_ROUTE => { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct Prepare { + content_id: String, + #[serde(default)] + retry: bool, + } + let input: Prepare = serde_json::from_slice(&bytes)?; + let identity = std::sync::Arc::new( + NodeIdentity::load_existing(&data_dir.join("identity")).await?, + ); + anyhow::ensure!(identity.did_key()? == audience, "Node identity changed"); + let root = data_dir.clone(); + serde_json::to_value( + tokio::task::spawn_blocking(move || { + crate::registered_media::prepare_registered( + root, + identity, + &input.content_id, + input.retry, + ) + }) + .await??, + )? + } protocol::OFFER_ROUTE => { let body: OfferRequest = serde_json::from_slice(&bytes)?; anyhow::ensure!( diff --git a/core/archipelago/src/api/handler/registered_media.rs b/core/archipelago/src/api/handler/registered_media.rs index 09e36572..6a9a8b72 100644 --- a/core/archipelago/src/api/handler/registered_media.rs +++ b/core/archipelago/src/api/handler/registered_media.rs @@ -4,7 +4,6 @@ use crate::{content_server::ByteRange, identity::NodeIdentity, registered_media: use anyhow::{Context, Result}; use hyper::{Body, HeaderMap, Response, StatusCode}; use std::sync::Arc; -use tokio::io::{AsyncReadExt, AsyncSeekExt}; fn route(path: &str) -> Result<(&str, &str)> { let (content, purchase) = path @@ -46,6 +45,125 @@ fn denied(message: &'static str) -> Response { } impl ApiHandler { + pub(super) async fn handle_rental_control( + &self, + mut request: hyper::Request, + ) -> Result> { + use hyper::body::HttpBody; + #[derive(serde::Deserialize)] + #[serde(deny_unknown_fields)] + struct Control { + capability: String, + ready_id: Option, + #[serde(default)] + retry: bool, + } + let path = request.uri().path().to_owned(); + let (base, action) = path.rsplit_once('/').context("Invalid rental action")?; + anyhow::ensure!( + matches!(action, "prepare" | "start") && request.method() == hyper::Method::POST, + "Invalid rental action" + ); + let (content, purchase) = route(base)?; + let bytes = tokio::time::timeout(std::time::Duration::from_secs(15), async { + let mut bytes = Vec::new(); + while let Some(chunk) = request.body_mut().data().await { + let chunk = chunk?; + anyhow::ensure!( + bytes + .len() + .checked_add(chunk.len()) + .is_some_and(|n| n <= 16 * 1024), + "Rental request too large" + ); + bytes.extend_from_slice(&chunk); + } + Ok::<_, anyhow::Error>(bytes) + }) + .await + .context("Rental request timed out")??; + let audience = crate::identity::did_key_from_pubkey_hex(&self.self_pubkey_hex)?; + let buyer = crate::content_auth::authenticate_request( + request.headers(), + &audience, + &hyper::Method::POST, + &path, + &bytes, + chrono::Utc::now().timestamp(), + )?; + let input: Control = serde_json::from_slice(&bytes)?; + let identity = + Arc::new(NodeIdentity::load_existing(&self.config.data_dir.join("identity")).await?); + anyhow::ensure!(identity.did_key()? == audience, "Node identity changed"); + let result = if action == "prepare" { + anyhow::ensure!(input.ready_id.is_none(), "Prepare does not start a rental"); + let prior = crate::registered_media::paid_window( + &self.config.data_dir, + &identity, + content, + purchase, + &buyer, + &input.capability, + ) + .await?; + let metadata = crate::registered_media::registered_metadata( + &self.config.data_dir, + &identity, + content, + )?; + anyhow::ensure!( + prior + .as_ref() + .is_none_or(|window| clock() >= window.started_at), + "Rental clock moved backwards" + ); + if let Some(window) = prior.as_ref().filter(|window| clock() >= window.expires_at) { + serde_json::json!({"state":"expired", "viewing_seconds":metadata.0.viewing_seconds,"started_at":window.started_at,"expires_at":window.expires_at}) + } else { + let state = crate::registered_media::prepare_paid( + self.config.data_dir.clone(), + identity, + content.into(), + purchase.into(), + buyer, + input.capability, + input.retry, + ) + .await?; + let mut result = serde_json::to_value(state)?; + result["viewing_seconds"] = serde_json::json!(metadata.0.viewing_seconds); + result["started_at"] = + serde_json::json!(prior.as_ref().map(|window| window.started_at)); + result["expires_at"] = + serde_json::json!(prior.as_ref().map(|window| window.expires_at)); + result + } + } else { + anyhow::ensure!(!input.retry, "Start cannot retry verification"); + let ready_id = input + .ready_id + .context("Media must be ready before explicit Start")?; + let window = crate::registered_media::start_paid( + self.config.data_dir.clone(), + identity, + content.into(), + purchase.into(), + buyer, + input.capability, + ready_id, + ) + .await?; + anyhow::ensure!(clock() >= window.started_at, "Rental clock moved backwards"); + serde_json::json!({"state": if clock() >= window.expires_at {"expired"} else {"started"}, + "started_at":window.started_at,"expires_at":window.expires_at}) + }; + Ok(build_response( + StatusCode::OK, + "application/json", + Body::from(serde_json::to_vec(&result)?), + )) + } + pub(super) async fn handle_registered_rental( &self, path: &str, @@ -102,7 +220,7 @@ impl ApiHandler { let selected = content.to_owned(); let key = identity.clone(); let metadata = tokio::task::spawn_blocking(move || { - crate::registered_media::registered_terms(&data, &key, &selected) + crate::registered_media::registered_metadata(&data, &key, &selected) }) .await?; let (receipt, _) = match metadata { @@ -160,11 +278,9 @@ async fn rental_response( let started = opened.started_at; let expires = opened.expires_at; let (start, length) = range.map_or((0, total), |(start, end)| (start, end - start + 1)); - let mut file = tokio::fs::File::from_std(opened.file); - file.seek(std::io::SeekFrom::Start(start)).await?; let chunks = futures_util::stream::try_unfold( - (file, length, now), - move |(mut file, left, now)| async move { + (opened.file, opened.verification, start, length, now), + move |(mut file, verification, position, left, now)| async move { if left == 0 { return Ok::<_, std::io::Error>(None); } @@ -175,15 +291,20 @@ async fn rental_response( "Rental window ended", )); } - let mut bytes = vec![0; left.min(64 * 1024) as usize]; - let count = tokio::time::timeout( - std::time::Duration::from_secs(expires - instant), - file.read(&mut bytes), - ) - .await - .map_err(|_| { - std::io::Error::new(std::io::ErrorKind::TimedOut, "Rental window ended") - })??; + let read = tokio::task::spawn_blocking(move || { + let bytes = verification + .index + .read_slice(&mut file, position, left.min(64 * 1024) as usize) + .map_err(std::io::Error::other)?; + Ok::<_, std::io::Error>((file, verification, bytes)) + }); + let (file, verification, bytes) = + tokio::time::timeout(std::time::Duration::from_secs(expires - instant), read) + .await + .map_err(|_| { + std::io::Error::new(std::io::ErrorKind::TimedOut, "Rental window ended") + })? + .map_err(std::io::Error::other)??; let instant = now(); if instant < started || instant >= expires { return Err(std::io::Error::new( @@ -191,14 +312,11 @@ async fn rental_response( "Rental window ended", )); } - if count == 0 { - return Err(std::io::Error::new( - std::io::ErrorKind::UnexpectedEof, - "Registered snapshot ended early", - )); - } - bytes.truncate(count); - Ok(Some((bytes, (file, left - count as u64, now)))) + let count = bytes.len() as u64; + Ok(Some(( + bytes, + (file, verification, position + count, left - count, now), + ))) }, ); let mut response = Response::builder() @@ -257,10 +375,19 @@ mod tests { ); } fn opened(size: u64) -> OpenedMedia { - let file = tempfile::tempfile().unwrap(); + use sha2::{Digest, Sha256}; + let mut file = tempfile::tempfile().unwrap(); file.set_len(size).unwrap(); + let binding = crate::rental_chunk_index::Binding { + content_id: format!("registered_{}", uuid::Uuid::new_v4()), + receipt_sha256: "ab".repeat(32), + full_sha256: hex::encode(Sha256::digest(vec![0; size as usize])), + size, + }; + let index = crate::rental_chunk_index::Index::scan(&mut file, binding, |_| Ok(())).unwrap(); OpenedMedia { file, + verification: crate::rental_readiness::Ready::fixture(index), size_bytes: size, mime_type: "video/mp4".into(), started_at: 1000, diff --git a/core/archipelago/src/api/rpc/dispatcher.rs b/core/archipelago/src/api/rpc/dispatcher.rs index e04bf941..f7f227f5 100644 --- a/core/archipelago/src/api/rpc/dispatcher.rs +++ b/core/archipelago/src/api/rpc/dispatcher.rs @@ -333,6 +333,8 @@ impl RpcHandler { "content.indeehub-projects" => self.handle_content_indeehub_projects().await, "content.browse-all-peers" => self.handle_content_browse_all_peers().await, "content.playback-handle" => self.handle_playback_handle(params, session_token).await, + "content.playback-prepare" => self.handle_playback_prepare(params, session_token).await, + "content.playback-start" => self.handle_playback_start(params, session_token).await, "content.playback-status" => self.handle_playback_status(params, session_token).await, "content.rental-purchase" => self.handle_content_rental_purchase(params).await, "content.purchase" => self.handle_content_purchase(params).await, diff --git a/core/archipelago/src/api/rpc/mod.rs b/core/archipelago/src/api/rpc/mod.rs index a576b095..e5c04f17 100644 --- a/core/archipelago/src/api/rpc/mod.rs +++ b/core/archipelago/src/api/rpc/mod.rs @@ -20,8 +20,8 @@ mod interfaces; pub(crate) mod lnd; mod marketplace; mod media_registration; -mod purchase; mod playback; +mod purchase; // pub(crate): 13-10's `assistant::backends::select_backend` reuses // `mesh::assistant::detect_ollama()` (D-04) rather than re-probing — // matches the existing `pub(crate) mod bitcoin_relay;`/`pub(crate) mod @@ -113,6 +113,8 @@ fn native_consent_origin_allowed(method: &str, headers: &hyper::HeaderMap, dev_m | "content.cancel-purchase" | "content.playback-handle" | "content.playback-status" + | "content.playback-prepare" + | "content.playback-start" ) || nostr_signing_origin_allowed(headers, dev_mode) } @@ -831,6 +833,8 @@ mod nostr_signing_origin_tests { "content.cancel-purchase", "content.playback-handle", "content.playback-status", + "content.playback-prepare", + "content.playback-start", ] { assert!(!native_consent_origin_allowed( method, diff --git a/core/archipelago/src/api/rpc/playback.rs b/core/archipelago/src/api/rpc/playback.rs index 010da04e..d8d69ceb 100644 --- a/core/archipelago/src/api/rpc/playback.rs +++ b/core/archipelago/src/api/rpc/playback.rs @@ -55,9 +55,7 @@ impl RpcHandler { let expires = self .playback_handles() .expiry(&handle, &session, &context)?; - Ok( - serde_json::json!({"playback_url":format!("/api/rental-playback/{handle}"), "expires_at":expires}), - ) + Ok(serde_json::json!({"handle":handle,"expires_at":expires})) } pub(super) async fn handle_playback_status( &self, @@ -72,3 +70,179 @@ impl RpcHandler { Ok(serde_json::json!({"expires_at":expires})) } } + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Prepare { + handle: String, + #[serde(default)] + retry: bool, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Start { + handle: String, + ready_id: String, +} +impl RpcHandler { + async fn playback_control( + &self, + handle: &str, + ready_id: Option<&str>, + retry: bool, + session: &Option, + ) -> Result { + let (session, context) = self.playback_context(session).await?; + let binding = self.playback_handles().lookup(handle, &session, &context)?; + let (capability, duration) = { + let journal = crate::content_purchase::Journal::open(&self.config.data_dir).await?; + let record = journal + .buyer(&binding.contract.id) + .await? + .context("Original purchase missing")?; + anyhow::ensure!( + record.contract == binding.contract, + "Original purchase changed" + ); + let capability = record + .receipt() + .context("Original payment is not settled")? + .capability + .clone(); + let envelope = journal + .protocol_envelope("buyer", &binding.contract.id) + .await? + .context("Original rental terms missing")?; + ( + capability, + envelope + .offer + .viewing_seconds + .context("Purchase is not a timed rental")?, + ) + }; + let peer = crate::federation::load_unique_payment_peer( + &self.config.data_dir, + &binding.seller_onion, + ) + .await?; + anyhow::ensure!( + peer.did == binding.contract.seller_did, + "Purchased seller identity changed" + ); + let mesh = peer.fips_npub.context("Seller mesh binding unavailable")?; + let action = if ready_id.is_some() { + "start" + } else { + "prepare" + }; + let path = format!( + "/content/{}/rental/{}/{action}", + binding.contract.content_id, binding.contract.id + ); + let body = serde_json::json!({"capability":capability,"ready_id":ready_id,"retry":retry}); + let (mut response, _) = + crate::fips::dial::PeerRequest::new(Some(&mesh), &binding.seller_onion, &path) + .require_fips() + .single_delivery() + .timeout(std::time::Duration::from_secs(20)) + .send_content_json(&self.config.data_dir, &binding.contract.seller_did, &body) + .await?; + anyhow::ensure!( + response.status().is_success(), + "Original rental is unavailable; retry this purchase without paying again" + ); + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await? { + anyhow::ensure!( + bytes + .len() + .checked_add(chunk.len()) + .is_some_and(|n| n <= 16 * 1024), + "Rental control response too large" + ); + bytes.extend_from_slice(&chunk); + } + let remote: serde_json::Value = serde_json::from_slice(&bytes)?; + let state = remote["state"].as_str().context("Missing rental state")?; + let window = match (remote["started_at"].as_u64(), remote["expires_at"].as_u64()) { + (None, None) if remote["started_at"].is_null() && remote["expires_at"].is_null() => { + None + } + (Some(started), Some(expires)) + if started > 0 && started.checked_add(duration) == Some(expires) => + { + Some((started, expires)) + } + _ => anyhow::bail!("Seller changed the original rental window"), + }; + let (started_at, expires_at) = + window.map_or((None, None), |(start, end)| (Some(start), Some(end))); + let result = match state { + "preparing" if ready_id.is_none() => { + let completed = remote["completed_bytes"] + .as_u64() + .context("Invalid verification progress")?; + anyhow::ensure!( + remote["total_bytes"].as_u64() == Some(binding.contract.content_size) + && completed <= binding.contract.content_size + && remote["viewing_seconds"].as_u64() == Some(duration), + "Rental preparation terms changed" + ); + serde_json::json!({"state":"preparing","completed_bytes":completed,"total_bytes":binding.contract.content_size, + "viewing_seconds":duration,"started_at":started_at,"expires_at":expires_at}) + } + "ready" if ready_id.is_none() => { + let id = remote["ready_id"] + .as_str() + .context("Missing readiness identifier")?; + let parsed = uuid::Uuid::parse_str(id)?; + anyhow::ensure!( + parsed.to_string() == id + && parsed.get_version_num() == 4 + && remote["total_bytes"].as_u64() == Some(binding.contract.content_size) + && remote["viewing_seconds"].as_u64() == Some(duration), + "Rental readiness changed" + ); + serde_json::json!({"state":"ready","ready_id":id,"handle":handle,"viewing_seconds":duration, + "started_at":started_at,"expires_at":expires_at}) + } + "unavailable" if ready_id.is_none() => { + serde_json::json!({"state":"unavailable","expires_at":expires_at}) + } + "expired" if window.is_some() => { + serde_json::json!({"state":"expired","started_at":started_at,"expires_at":expires_at}) + } + "started" if ready_id.is_some() && window.is_some() => { + serde_json::json!({"state":"started","started_at":started_at, + "expires_at":expires_at,"playback_url":format!("/api/rental-playback/{handle}")}) + } + _ => { + anyhow::bail!("Unexpected rental state; recover this purchase without paying again") + } + }; + if let Some((_, expires)) = window { + self.playback_handles() + .note_expiry(handle, &binding, expires)?; + } + Ok(result) + } + pub(super) async fn handle_playback_prepare( + &self, + params: Option, + session: &Option, + ) -> Result { + let input: Prepare = serde_json::from_value(params.context("Missing playback handle")?)?; + self.playback_control(&input.handle, None, input.retry, session) + .await + } + pub(super) async fn handle_playback_start( + &self, + params: Option, + session: &Option, + ) -> Result { + let input: Start = serde_json::from_value(params.context("Missing readiness identifier")?)?; + self.playback_control(&input.handle, Some(&input.ready_id), false, session) + .await + } +} diff --git a/core/archipelago/src/api/rpc/purchase.rs b/core/archipelago/src/api/rpc/purchase.rs index 1a613d47..da51aea9 100644 --- a/core/archipelago/src/api/rpc/purchase.rs +++ b/core/archipelago/src/api/rpc/purchase.rs @@ -51,6 +51,9 @@ impl RpcHandler { ) .await?; match result { + ReadyPurchase::Preparing { .. } => { + anyhow::bail!("Ordinary purchase cannot prepare a timed rental") + } ReadyPurchase::AwaitingConfirmation { operation_id, envelope_sha256, @@ -127,6 +130,8 @@ impl RpcHandler { #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct RentalParams { + #[serde(default)] + retry_preparation: bool, seller_did: String, content_id: String, expected_sha256: String, @@ -168,8 +173,9 @@ impl RpcHandler { ¶ms.seller_did, ) .await?; - let transport = - FipsPurchaseTransport::load(self.config.data_dir.clone(), onion.clone()).await?; + let transport = FipsPurchaseTransport::load(self.config.data_dir.clone(), onion.clone()) + .await? + .retry_preparation(params.retry_preparation); let expected = caller::ExpectedRental { seller_did: params.seller_did, content_id: params.content_id.clone(), @@ -189,6 +195,11 @@ impl RpcHandler { ) .await? { + ReadyPurchase::Preparing { + completed_bytes, + total_bytes, + } => Ok(serde_json::json!({ + "state":"preparing", "completed_bytes":completed_bytes,"total_bytes":total_bytes})), ReadyPurchase::AwaitingConfirmation { operation_id, envelope_sha256, diff --git a/core/archipelago/src/content_purchase_caller.rs b/core/archipelago/src/content_purchase_caller.rs index 09d85154..d8249b62 100644 --- a/core/archipelago/src/content_purchase_caller.rs +++ b/core/archipelago/src/content_purchase_caller.rs @@ -12,6 +12,12 @@ pub(crate) trait PurchaseTransport: Send + Sync { /// This identity is the independently verified peer binding, not response JSON. fn seller_did(&self) -> &str; fn seller_onion(&self) -> &str; + fn prepare_offer( + &self, + _content_id: &str, + ) -> impl Future>> + Send { + async { Ok(None) } + } fn offer(&self, id: &str, content_id: &str) -> impl Future> + Send; fn accept(&self, envelope: &Envelope) -> impl Future> + Send; fn status(&self, envelope: &Envelope) -> impl Future> + Send; @@ -20,6 +26,10 @@ pub(crate) trait PurchaseTransport: Send + Sync { } #[derive(Clone)] pub(crate) enum ReadyPurchase { + Preparing { + completed_bytes: u64, + total_bytes: u64, + }, AwaitingConfirmation { operation_id: String, envelope_sha256: String, @@ -210,6 +220,16 @@ pub(crate) async fn purchase_bound( !unresolved_owned, "Prior purchase delivery needs recovery; no new payment operation was created" ); + if content_id.starts_with("registered_") { + if let Some((completed_bytes, total_bytes)) = + transport.prepare_offer(content_id).await? + { + return Ok(ReadyPurchase::Preparing { + completed_bytes, + total_bytes, + }); + } + } let id = uuid::Uuid::new_v4().to_string(); let offer = transport.offer(&id, content_id).await?; offer.validate()?; diff --git a/core/archipelago/src/content_purchase_protocol.rs b/core/archipelago/src/content_purchase_protocol.rs index f5377e49..7ea4d207 100644 --- a/core/archipelago/src/content_purchase_protocol.rs +++ b/core/archipelago/src/content_purchase_protocol.rs @@ -8,6 +8,7 @@ use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use std::path::Path; +pub(crate) const PREPARE_OFFER_ROUTE: &str = "/content/purchase/v1/prepare-offer"; pub(crate) const OFFER_ROUTE: &str = "/content/purchase/v1/offer"; pub(crate) const ACCEPT_ROUTE: &str = "/content/purchase/v1/accept"; pub(crate) const SETTLE_ROUTE: &str = "/content/purchase/v1/settle"; diff --git a/core/archipelago/src/content_purchase_transport.rs b/core/archipelago/src/content_purchase_transport.rs index b689c9d0..1d6dfc27 100644 --- a/core/archipelago/src/content_purchase_transport.rs +++ b/core/archipelago/src/content_purchase_transport.rs @@ -15,6 +15,7 @@ pub(crate) struct FipsPurchaseTransport { onion: String, seller_did: String, fips_npub: String, + retry_preparation: bool, } impl FipsPurchaseTransport { pub async fn load(data_dir: PathBuf, onion: String) -> Result { @@ -31,8 +32,13 @@ impl FipsPurchaseTransport { onion, seller_did: peer.did, fips_npub, + retry_preparation: false, }) } + pub fn retry_preparation(mut self, retry: bool) -> Self { + self.retry_preparation = retry; + self + } async fn post( &self, route: &str, @@ -73,6 +79,30 @@ impl PurchaseTransport for FipsPurchaseTransport { fn seller_did(&self) -> &str { &self.seller_did } + async fn prepare_offer(&self, content_id: &str) -> Result> { + let state: crate::rental_readiness::Status = self + .post( + protocol::PREPARE_OFFER_ROUTE, + &serde_json::json!({"content_id":content_id,"retry":self.retry_preparation}), + ) + .await?; + match state { + crate::rental_readiness::Status::Ready { .. } => Ok(None), + crate::rental_readiness::Status::Preparing { + completed_bytes, + total_bytes, + } => { + anyhow::ensure!( + completed_bytes <= total_bytes && total_bytes > 0, + "Invalid media preparation progress" + ); + Ok(Some((completed_bytes, total_bytes))) + } + crate::rental_readiness::Status::Unavailable => anyhow::bail!( + "Media verification failed; no payment started. Retry media preparation." + ), + } + } async fn offer(&self, id: &str, content_id: &str) -> Result { #[derive(Serialize)] struct Request<'a> { diff --git a/core/archipelago/src/main.rs b/core/archipelago/src/main.rs index dc12616a..314937c3 100644 --- a/core/archipelago/src/main.rs +++ b/core/archipelago/src/main.rs @@ -83,6 +83,8 @@ mod port_allocator; mod prepared_media; mod rate_limit; mod registered_media; +mod rental_chunk_index; +mod rental_readiness; pub mod seed; mod server; mod session; diff --git a/core/archipelago/src/registered_media.rs b/core/archipelago/src/registered_media.rs index 0299d789..bec98429 100644 --- a/core/archipelago/src/registered_media.rs +++ b/core/archipelago/src/registered_media.rs @@ -86,6 +86,7 @@ struct Lease { /// stream at expiry, and must not call open_paid for HTTP HEAD/preflight. pub(crate) struct OpenedMedia { pub file: File, + pub verification: Arc, pub size_bytes: u64, pub mime_type: String, pub started_at: u64, @@ -395,6 +396,13 @@ fn verify( Ok(()) } fn open_snapshot(data_dir: &Path, record: &Registered) -> Result { + open_snapshot_with_stamp(data_dir, record, true) +} +fn open_snapshot_with_stamp( + data_dir: &Path, + record: &Registered, + require_original_stamp: bool, +) -> Result { use std::ffi::CString; use std::os::fd::FromRawFd; // Resolve from the configured data directory in one kernel operation. No @@ -435,7 +443,8 @@ fn open_snapshot(data_dir: &Path, record: &Registered) -> Result { ); let file = unsafe { File::from_raw_fd(fd as i32) }; anyhow::ensure!( - Stamp::from_file(&file)? == record.stamp, + Stamp::from_file(&file)?.size == record.stamp.size + && (!require_original_stamp || Stamp::from_file(&file)? == record.stamp), "Registered immutable snapshot changed" ); Ok(file) @@ -561,11 +570,11 @@ pub(crate) fn resolve_registration( /// No request chooses app scope or storage path. This is an offer prerequisite, /// not advertisement: the future offer creator must authenticate the peer and /// bind all returned terms into the purchase contract before seller acceptance. -pub(crate) fn registered_terms( +fn metadata_record( data_dir: &Path, identity: &NodeIdentity, content_id: &str, -) -> Result<(Receipt, String)> { +) -> Result { let id = registration_id(content_id)?; let pin = registration_pin::load_existing(data_dir, APP_ID, identity)?; let held = held(data_dir, false)?; @@ -573,32 +582,198 @@ pub(crate) fn registered_terms( .context("Registered content is unavailable")?; drop(held); verify(&record, &pin, identity)?; - let mut file = open_snapshot(data_dir, &record)?; - // A quote must not invite payment for altered bytes, including a same-tick - // metadata collision. This scan completes before any offer is accepted. - ensure_verified_for_use(data_dir, &record, &mut file, true)?; + Ok(record) +} +fn chunk_binding(record: &Registered) -> Result { + Ok(crate::rental_chunk_index::Binding { + content_id: record.receipt.content_id.clone(), + receipt_sha256: hash(&serde_json::to_vec(&record.receipt)?), + full_sha256: record.receipt.sha256.clone(), + size: record.stamp.size, + }) +} +fn readiness_key(data_dir: &Path, record: &Registered) -> Result { + Ok(format!( + "{}:{}", + data_dir.canonicalize()?.display(), + hash(&serde_json::to_vec(&record.receipt)?) + )) +} +/// Signed metadata only: no whole-file scan or lease is permitted here. +pub(crate) fn registered_metadata( + data_dir: &Path, + identity: &NodeIdentity, + content_id: &str, +) -> Result<(Receipt, String)> { + let record = metadata_record(data_dir, identity, content_id)?; + Ok((record.receipt, record.terms_sha256)) +} +/// Starts/reuses bounded background verification, before quoting or playing. +/// The signed index survives restart; range requests never invoke this scan. +pub(crate) fn prepare_registered( + data_dir: PathBuf, + identity: Arc, + content_id: &str, + retry: bool, +) -> Result { + let record = metadata_record(&data_dir, &identity, content_id)?; + let binding = chunk_binding(&record)?; + let key = readiness_key(&data_dir, &record)?; + let manager = crate::rental_readiness::shared(); + if retry { + manager.retry_failed(&key, &binding)?; + } + manager.prepare(key, binding.clone(), move |progress| { + let held = keyed(&data_dir, "verify", &record.receipt.request_id)?; + let path = held + .path + .join(format!("chunks-{}.bin", record.receipt.request_id)); + let mut file = open_snapshot_with_stamp(&data_dir, &record, false)?; + if let Ok(Some(index)) = crate::rental_chunk_index::Index::load(&path, &binding, &identity) + { + return Ok(index); + } + // A damaged derived index is not evidence about the original bytes. + // Rebuild only after a complete scan matches the signed producer SHA; + // this avoids making an editable cache a permanent recovery dead end. + let index = crate::rental_chunk_index::Index::scan(&mut file, binding, |n| { + progress.store(n, std::sync::atomic::Ordering::SeqCst); + Ok(()) + })?; + index.save(&path, &identity)?; + Ok(index) + }) +} +#[derive(Clone, Serialize)] +pub(crate) struct RentalWindow { + pub started_at: u64, + pub expires_at: u64, +} +fn verify_lease( + lease: &Lease, + contract: &Contract, + capability: &str, + duration: u64, +) -> Result { + anyhow::ensure!( + lease.version == 1 + && lease.purchase_id == contract.id + && lease.buyer_did == contract.buyer_did + && lease.content_id == contract.content_id + && lease.contract_hash == contract.context_hash()? + && lease.capability_hash == hash(capability.as_bytes()) + && lease.started_at.checked_add(duration) == Some(lease.expires_at), + "Persisted rental terms changed" + ); + Ok(RentalWindow { + started_at: lease.started_at, + expires_at: lease.expires_at, + }) +} +fn verify_settled_terms(record: &Registered, contract: &Contract) -> Result<()> { + anyhow::ensure!( + contract.content_sha256 == record.receipt.sha256 + && contract.content_size == record.stamp.size + && contract.terms_sha256 == record.terms_sha256 + && record.receipt.price_sats > 0 + && record + .receipt + .payment_methods + .iter() + .any(|method| method == "cashu") + && contract.minimum_net_sats == record.receipt.price_sats, + "Settled purchase does not match registered immutable terms" + ); + Ok(()) +} +/// Only explicit authenticated Start invokes this commit. Losing its response +/// cannot reset the original clock; repeated Start returns the same window. +fn start_ready( + data_dir: &Path, + identity: &NodeIdentity, + contract: &Contract, + capability: &str, + ready_id: &str, + now: impl Fn() -> Result, +) -> Result { + let record = metadata_record(data_dir, identity, &contract.content_id)?; + verify_settled_terms(&record, contract)?; + { + let held = keyed(data_dir, "lease", &contract.id)?; + if let Some(lease) = read::(&held.path.join(format!("lease-{}.json", contract.id)))? + { + return verify_lease(&lease, contract, capability, record.receipt.viewing_seconds); + } + } + let ready = crate::rental_readiness::shared().ready( + &readiness_key(data_dir, &record)?, + &chunk_binding(&record)?, + Some(ready_id), + )?; + let mut file = open_snapshot_with_stamp(data_dir, &record, false)?; + // A bounded actual byte read before the first clock. Later chunks are + // verified immediately before delivery, including seek/range requests. + ready.index.read_slice(&mut file, 0, 1)?; + let held = keyed(data_dir, "lease", &contract.id)?; + let path = held.path.join(format!("lease-{}.json", contract.id)); + if let Some(lease) = read::(&path)? { + return verify_lease(&lease, contract, capability, record.receipt.viewing_seconds); + } + let started_at = now()?; + let expires_at = started_at + .checked_add(record.receipt.viewing_seconds) + .context("Rental expiry overflow")?; + let lease = Lease { + version: 1, + purchase_id: contract.id.clone(), + buyer_did: contract.buyer_did.clone(), + content_id: contract.content_id.clone(), + contract_hash: contract.context_hash()?, + capability_hash: hash(capability.as_bytes()), + started_at, + expires_at, + }; + persist(&held, &path, &lease)?; + Ok(RentalWindow { + started_at, + expires_at, + }) +} + +pub(crate) fn registered_terms( + data_dir: &Path, + identity: &NodeIdentity, + content_id: &str, +) -> Result<(Receipt, String)> { + let record = metadata_record(data_dir, identity, content_id)?; + let _ready = crate::rental_readiness::shared().ready( + &readiness_key(data_dir, &record)?, + &chunk_binding(&record)?, + None, + )?; + let _file = open_snapshot_with_stamp(data_dir, &record, false)?; Ok((record.receipt, record.terms_sha256)) } /// Only authenticated peer GET/range routes may call this. The capability is /// checked against this node's durable seller journal; client receipts do not /// establish payment. A lease is persisted before any bytes can be returned. -pub(crate) async fn open_paid( - data_dir: PathBuf, - identity: Arc, - content_id: String, - purchase_id: String, - authenticated_buyer: String, - capability: String, -) -> Result { +async fn paid_contract( + data_dir: &Path, + identity: &NodeIdentity, + content_id: &str, + purchase_id: &str, + authenticated_buyer: &str, + capability: &str, +) -> Result { uuid(&purchase_id)?; registration_id(&content_id)?; anyhow::ensure!( capability.len() == 64 && capability.bytes().all(|c| c.is_ascii_hexdigit()), "Invalid delivery capability" ); - let (contract, saved_capability) = { - let journal = Journal::open(&data_dir).await?; + let contract = { + let journal = Journal::open(data_dir).await?; let record = journal .seller(&purchase_id) .await? @@ -614,18 +789,157 @@ pub(crate) async fn open_paid( _ => anyhow::bail!("Seller receipt is not durable"), }; anyhow::ensure!( - constant_equal(&capability, &receipt.capability), + constant_equal(capability, &receipt.capability), "Delivery capability does not match this purchase" ); - (record.contract, receipt.capability) + record.contract }; + Ok(contract) +} +pub(crate) async fn prepare_paid( + data_dir: PathBuf, + identity: Arc, + content_id: String, + purchase_id: String, + buyer: String, + capability: String, + retry: bool, +) -> Result { + paid_contract( + &data_dir, + &identity, + &content_id, + &purchase_id, + &buyer, + &capability, + ) + .await?; + tokio::task::spawn_blocking(move || prepare_registered(data_dir, identity, &content_id, retry)) + .await? +} +pub(crate) async fn paid_window( + data_dir: &Path, + identity: &NodeIdentity, + content_id: &str, + purchase_id: &str, + buyer: &str, + capability: &str, +) -> Result> { + let contract = paid_contract( + data_dir, + identity, + content_id, + purchase_id, + buyer, + capability, + ) + .await?; + let record = metadata_record(data_dir, identity, content_id)?; + verify_settled_terms(&record, &contract)?; + let held = keyed(data_dir, "lease", purchase_id)?; + read::(&held.path.join(format!("lease-{purchase_id}.json")))? + .map(|lease| { + verify_lease( + &lease, + &contract, + capability, + record.receipt.viewing_seconds, + ) + }) + .transpose() +} +pub(crate) async fn start_paid( + data_dir: PathBuf, + identity: Arc, + content_id: String, + purchase_id: String, + buyer: String, + capability: String, + ready_id: String, +) -> Result { + let contract = paid_contract( + &data_dir, + &identity, + &content_id, + &purchase_id, + &buyer, + &capability, + ) + .await?; tokio::task::spawn_blocking(move || { - open_settled(&data_dir, &identity, &contract, &saved_capability, || { + start_ready( + &data_dir, + &identity, + &contract, + &capability, + &ready_id, + || u64::try_from(chrono::Utc::now().timestamp()).context("Invalid node clock"), + ) + }) + .await? +} +pub(crate) async fn open_paid( + data_dir: PathBuf, + identity: Arc, + content_id: String, + purchase_id: String, + authenticated_buyer: String, + capability: String, +) -> Result { + let contract = paid_contract( + &data_dir, + &identity, + &content_id, + &purchase_id, + &authenticated_buyer, + &capability, + ) + .await?; + tokio::task::spawn_blocking(move || { + open_started(&data_dir, &identity, &contract, &capability, || { u64::try_from(chrono::Utc::now().timestamp()).context("Invalid node clock") }) }) .await? } +fn open_started( + data_dir: &Path, + identity: &NodeIdentity, + contract: &Contract, + capability: &str, + now: impl Fn() -> Result, +) -> Result { + let record = metadata_record(data_dir, identity, &contract.content_id)?; + verify_settled_terms(&record, contract)?; + let lease = { + let held = keyed(data_dir, "lease", &contract.id)?; + read::(&held.path.join(format!("lease-{}.json", contract.id)))? + .context("Explicit Start is required; no rental was started")? + }; + let window = verify_lease(&lease, contract, capability, record.receipt.viewing_seconds)?; + let instant = now()?; + anyhow::ensure!( + instant >= window.started_at && instant < window.expires_at, + "Rental expired or node clock moved backwards" + ); + let verification = crate::rental_readiness::shared().ready( + &readiness_key(data_dir, &record)?, + &chunk_binding(&record)?, + None, + )?; + let file = open_snapshot_with_stamp(data_dir, &record, false)?; + Ok(OpenedMedia { + file, + verification, + size_bytes: record.stamp.size, + mime_type: record.mime_type, + started_at: window.started_at, + expires_at: window.expires_at, + }) +} +// Existing clock/receipt invariant fixtures exercise an explicit local start; +// public prepare/start/open separation is tested independently below. +#[cfg(test)] fn open_settled( data_dir: &Path, identity: &NodeIdentity, @@ -701,8 +1015,12 @@ fn open_settled( now >= lease.started_at && now < lease.expires_at, "Rental expired or node clock moved backwards" ); + let index = + crate::rental_chunk_index::Index::scan(&mut file, chunk_binding(&record)?, |_| Ok(()))?; + let verification = crate::rental_readiness::Ready::fixture(index); Ok(OpenedMedia { file, + verification, size_bytes: record.stamp.size, mime_type: record.mime_type, started_at: lease.started_at, @@ -786,6 +1104,30 @@ mod tests { |_| Ok(()), ) } + async fn ready(&self, receipt: &Receipt) -> String { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + match prepare_registered( + self.root.path().into(), + self.identity.clone(), + &receipt.content_id, + false, + ) + .unwrap() + { + crate::rental_readiness::Status::Ready { ready_id, .. } => return ready_id, + crate::rental_readiness::Status::Preparing { .. } => { + tokio::time::sleep(std::time::Duration::from_millis(5)).await + } + crate::rental_readiness::Status::Unavailable => { + panic!("fixture verification failed") + } + } + } + }) + .await + .expect("bounded fixture preparation") + } fn contract(&self, receipt: &Receipt) -> Contract { Contract { version: 1, @@ -872,6 +1214,7 @@ mod tests { std::fs::remove_file(fixture.root.path().join("cloud/film.mp4")).unwrap(); assert_eq!(fixture.register(2000).unwrap(), receipt); assert_eq!(std::fs::read(&mapping).unwrap(), original); + fixture.ready(&receipt).await; let found = registered_terms(fixture.root.path(), &fixture.identity, &receipt.content_id).unwrap(); assert_eq!(found, (receipt.clone(), terms(&receipt).unwrap())); @@ -1029,6 +1372,57 @@ mod tests { .join(STORE) .join(format!("lease-{}.json", contract.id)) .exists()); + let ready_id = fixture.ready(&receipt).await; + // Preparation, settlement and a GET are not permission to start the clock. + assert!(open_paid( + fixture.root.path().into(), + fixture.identity.clone(), + receipt.content_id.clone(), + contract.id.clone(), + contract.buyer_did.clone(), + settled.capability.clone() + ) + .await + .is_err()); + assert!(paid_window( + fixture.root.path(), + &fixture.identity, + &receipt.content_id, + &contract.id, + &contract.buyer_did, + &settled.capability + ) + .await + .unwrap() + .is_none()); + let window = start_paid( + fixture.root.path().into(), + fixture.identity.clone(), + receipt.content_id.clone(), + contract.id.clone(), + contract.buyer_did.clone(), + settled.capability.clone(), + ready_id, + ) + .await + .unwrap(); + // Lost Start response: a retry replays the durable original window even + // if its ephemeral readiness identifier no longer exists after restart. + let replay = start_paid( + fixture.root.path().into(), + fixture.identity.clone(), + receipt.content_id.clone(), + contract.id.clone(), + contract.buyer_did.clone(), + settled.capability.clone(), + uuid::Uuid::new_v4().to_string(), + ) + .await + .unwrap(); + assert_eq!( + (window.started_at, window.expires_at), + (replay.started_at, replay.expires_at) + ); let opened = open_paid( fixture.root.path().into(), fixture.identity.clone(), @@ -1041,6 +1435,33 @@ mod tests { .unwrap(); assert_eq!(opened.expires_at - opened.started_at, 60); assert_eq!(opened.size_bytes, 24); + for offset in [0, 8, 23] { + let mut range = open_paid( + fixture.root.path().into(), + fixture.identity.clone(), + receipt.content_id.clone(), + contract.id.clone(), + contract.buyer_did.clone(), + settled.capability.clone(), + ) + .await + .unwrap(); + assert_eq!( + range + .verification + .index + .read_slice(&mut range.file, offset, 1) + .unwrap() + .len(), + 1 + ); + assert_eq!(range.started_at, window.started_at); + } + assert_eq!( + crate::rental_chunk_index::scans(&receipt.content_id), + 1, + "quote preparation and repeated GET/ranges share one original SHA scan" + ); } #[tokio::test] async fn altered_snapshot_or_overflow_never_creates_first_rental() { @@ -1249,21 +1670,69 @@ mod tests { .exists()); } #[tokio::test] - async fn offer_preflight_rebuilds_verified_cache_without_creating_rental_and_rejects_corruption( - ) { + async fn signed_index_preparation_is_lease_free_and_unsigned_cache_cannot_change_it() { let fixture = Fixture::new().await; let receipt = fixture.register(1100).unwrap(); - let path = fixture + assert!( + registered_terms(fixture.root.path(), &fixture.identity, &receipt.content_id).is_err() + ); + let ready_id = fixture.ready(&receipt).await; + let record = + metadata_record(fixture.root.path(), &fixture.identity, &receipt.content_id).unwrap(); + let index_path = fixture + .root + .path() + .join(STORE) + .join(format!("chunks-{}.bin", receipt.request_id)); + let index = crate::rental_chunk_index::Index::load( + &index_path, + &chunk_binding(&record).unwrap(), + &fixture.identity, + ) + .unwrap() + .unwrap(); + assert_eq!(index.binding.full_sha256, receipt.sha256); + let cache = fixture .root .path() .join(STORE) .join(format!("verified-{}.json", receipt.request_id)); - let original = std::fs::read(&path).unwrap(); - std::fs::remove_file(&path).unwrap(); - let (terms, _) = - registered_terms(fixture.root.path(), &fixture.identity, &receipt.content_id).unwrap(); - assert_eq!(terms, receipt); - assert_eq!(std::fs::read(&path).unwrap(), original); + std::fs::write(cache, b"untrusted corrupt old metadata").unwrap(); + assert_eq!( + registered_terms(fixture.root.path(), &fixture.identity, &receipt.content_id) + .unwrap() + .0, + receipt + ); + assert_eq!(fixture.ready(&receipt).await, ready_id); + assert!(std::fs::read_dir(fixture.root.path().join(STORE)) + .unwrap() + .all(|entry| !entry + .unwrap() + .file_name() + .to_string_lossy() + .starts_with("lease-"))); + let mut bytes = std::fs::read(&index_path).unwrap(); + *bytes.last_mut().unwrap() ^= 1; + std::fs::write(&index_path, bytes).unwrap(); + assert!(crate::rental_chunk_index::Index::load( + &index_path, + &chunk_binding(&record).unwrap(), + &fixture.identity + ) + .is_err()); + crate::rental_readiness::shared() + .forget_for_restart(&readiness_key(fixture.root.path(), &record).unwrap()); + let replacement = fixture.ready(&receipt).await; + assert_ne!(replacement, ready_id); + assert_eq!(crate::rental_chunk_index::scans(&receipt.content_id), 2); + assert!(crate::rental_chunk_index::Index::load( + &index_path, + &chunk_binding(&record).unwrap(), + &fixture.identity + ) + .unwrap() + .is_some()); assert!(std::fs::read_dir(fixture.root.path().join(STORE)) .unwrap() .all(|entry| !entry @@ -1271,11 +1740,5 @@ mod tests { .file_name() .to_string_lossy() .starts_with("lease-"))); - let mut cache: VerifiedSnapshot = read(&path).unwrap().unwrap(); - cache.sha256 = "ff".repeat(32); - std::fs::write(&path, serde_json::to_vec(&cache).unwrap()).unwrap(); - assert!( - registered_terms(fixture.root.path(), &fixture.identity, &receipt.content_id).is_err() - ); } } diff --git a/core/archipelago/src/rental_chunk_index.rs b/core/archipelago/src/rental_chunk_index.rs new file mode 100644 index 00000000..fa624105 --- /dev/null +++ b/core/archipelago/src/rental_chunk_index.rs @@ -0,0 +1,319 @@ +//! Node-signed byte indexes, created only by a scan matching the producer's +//! original signed full SHA. Metadata alone never authorizes streamed bytes. +use crate::identity::NodeIdentity; +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{ + fs::{File, OpenOptions}, + io::{Read, Seek, SeekFrom, Write}, + os::unix::fs::OpenOptionsExt, + path::Path, +}; +pub(crate) const CHUNK_BYTES: usize = 64 * 1024; +const MAX_BYTES: u64 = 16 * 1024 * 1024 * 1024; +#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub(crate) struct Binding { + pub content_id: String, + pub receipt_sha256: String, + pub full_sha256: String, + pub size: u64, +} +impl Binding { + fn validate(&self) -> Result<()> { + anyhow::ensure!( + self.content_id + .strip_prefix("registered_") + .and_then(|id| uuid::Uuid::parse_str(id).ok()) + .map(|id| format!("registered_{id}") == self.content_id) + .unwrap_or(false) + && self.size > 0 + && self.size <= MAX_BYTES, + "Invalid byte index binding" + ); + for value in [&self.receipt_sha256, &self.full_sha256] { + anyhow::ensure!( + value.len() == 64 + && value + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)), + "Invalid byte index digest" + ); + } + Ok(()) + } + pub fn index_bytes(&self) -> Result { + self.validate()?; + Ok(self.size.div_ceil(CHUNK_BYTES as u64) * 32) + } +} +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Header { + version: u32, + binding: Binding, + chunks_sha256: String, + signature: String, +} +fn preimage(binding: &Binding, chunks_sha256: &str) -> Result> { + Ok(serde_json::to_vec(&( + "archipelago-rental-chunk-index-v1", + CHUNK_BYTES, + &binding.content_id, + &binding.receipt_sha256, + &binding.full_sha256, + binding.size, + chunks_sha256, + ))?) +} +pub(crate) struct Index { + pub binding: Binding, + chunks: Vec<[u8; 32]>, + pub commitment: String, +} +#[cfg(test)] +fn scan_counts() -> &'static std::sync::Mutex> { + static COUNTS: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + COUNTS.get_or_init(Default::default) +} +#[cfg(test)] +pub(crate) fn scans(content_id: &str) -> usize { + *scan_counts().lock().unwrap().get(content_id).unwrap_or(&0) +} +impl Index { + /// Exactly one whole-file scan. Caller runs it as a bounded background job; + /// progress/cancellation cannot create or alter a purchase lease. + pub fn scan( + file: &mut File, + binding: Binding, + mut progress: impl FnMut(u64) -> Result<()>, + ) -> Result { + let bytes = binding.index_bytes()?; + #[cfg(test)] + { + *scan_counts() + .lock() + .unwrap() + .entry(binding.content_id.clone()) + .or_default() += 1; + } + anyhow::ensure!( + file.metadata()?.is_file() && file.metadata()?.len() == binding.size, + "Snapshot size changed" + ); + file.seek(SeekFrom::Start(0))?; + let mut chunks = Vec::with_capacity(bytes as usize / 32); + let mut full = Sha256::new(); + let mut buffer = [0u8; CHUNK_BYTES]; + let mut read = 0; + while read < binding.size { + progress(read)?; + let count = (binding.size - read).min(CHUNK_BYTES as u64) as usize; + file.read_exact(&mut buffer[..count])?; + full.update(&buffer[..count]); + chunks.push(Sha256::digest(&buffer[..count]).into()); + read += count as u64; + } + anyhow::ensure!( + file.metadata()?.len() == binding.size + && hex::encode(full.finalize()) == binding.full_sha256, + "Registered snapshot content changed" + ); + progress(read)?; + let mut digest = Sha256::new(); + for chunk in &chunks { + digest.update(chunk); + } + let commitment = hex::encode(digest.finalize()); + file.seek(SeekFrom::Start(0))?; + Ok(Self { + binding, + chunks, + commitment, + }) + } + /// Persist under the registration's existing verification flock. Sync rename + /// is the commit point; no mutation is queued after a canceled future drops. + pub fn save(&self, path: &Path, identity: &NodeIdentity) -> Result<()> { + let header = Header { + version: 1, + binding: self.binding.clone(), + chunks_sha256: self.commitment.clone(), + signature: identity.sign(&preimage(&self.binding, &self.commitment)?), + }; + let encoded = serde_json::to_vec(&header)?; + anyhow::ensure!(encoded.len() <= 4096, "Byte index header too large"); + let parent = path.parent().context("Byte index directory missing")?; + let temporary = parent.join(format!(".chunk-index-{}.tmp", uuid::Uuid::new_v4())); + let result = (|| { + let mut file = OpenOptions::new() + .create_new(true) + .write(true) + .mode(0o600) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC) + .open(&temporary)?; + file.write_all(&(encoded.len() as u32).to_be_bytes())?; + file.write_all(&encoded)?; + for chunk in &self.chunks { + file.write_all(chunk)?; + } + file.sync_all()?; + std::fs::rename(&temporary, path)?; + File::open(parent)?.sync_all()?; + Ok(()) + })(); + if result.is_err() { + let _ = std::fs::remove_file(temporary); + } + result + } + /// A signed index survives restart without treating an editable metadata + /// cache as proof. Every delivered chunk is independently checked below. + pub fn load(path: &Path, binding: &Binding, identity: &NodeIdentity) -> Result> { + let expected = binding.index_bytes()?; + let mut file = match OpenOptions::new() + .read(true) + .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK) + .open(path) + { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + let metadata = file.metadata()?; + anyhow::ensure!( + metadata.is_file() && metadata.len() <= expected + 4100, + "Invalid byte index file" + ); + let mut len = [0u8; 4]; + file.read_exact(&mut len)?; + let len = u32::from_be_bytes(len) as usize; + anyhow::ensure!( + len <= 4096 && metadata.len() == 4 + len as u64 + expected, + "Invalid byte index length" + ); + let mut header = vec![0; len]; + file.read_exact(&mut header)?; + let header: Header = serde_json::from_slice(&header)?; + anyhow::ensure!( + header.version == 1 + && &header.binding == binding + && NodeIdentity::verify( + &identity.pubkey_hex(), + &preimage(binding, &header.chunks_sha256)?, + &header.signature + )?, + "Byte index signature or immutable binding changed" + ); + let mut chunks = Vec::with_capacity(expected as usize / 32); + let mut digest = Sha256::new(); + for _ in 0..expected / 32 { + let mut chunk = [0; 32]; + file.read_exact(&mut chunk)?; + digest.update(chunk); + chunks.push(chunk); + } + anyhow::ensure!( + hex::encode(digest.finalize()) == header.chunks_sha256, + "Byte index was altered" + ); + Ok(Some(Self { + binding: binding.clone(), + chunks, + commitment: header.chunks_sha256, + })) + } + /// Return only a slice of a completely verified aligned chunk. This bounds + /// seek/range work to64KiB and detects same-inode/same-stamp byte changes. + pub fn read_slice(&self, file: &mut File, start: u64, requested: usize) -> Result> { + anyhow::ensure!( + start < self.binding.size + && requested > 0 + && file.metadata()?.len() == self.binding.size, + "Snapshot range or size changed" + ); + let chunk = start / CHUNK_BYTES as u64; + let aligned = chunk * CHUNK_BYTES as u64; + let size = (self.binding.size - aligned).min(CHUNK_BYTES as u64) as usize; + let mut bytes = vec![0; size]; + file.seek(SeekFrom::Start(aligned))?; + file.read_exact(&mut bytes)?; + let actual: [u8; 32] = Sha256::digest(&bytes).into(); + anyhow::ensure!( + self.chunks.get(chunk as usize) == Some(&actual), + "Registered snapshot chunk changed; recover original entitlement" + ); + let offset = (start - aligned) as usize; + Ok(bytes[offset..offset + requested.min(size - offset)].to_vec()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + fn binding(bytes: &[u8]) -> Binding { + Binding { + content_id: format!("registered_{}", uuid::Uuid::new_v4()), + receipt_sha256: "ab".repeat(32), + full_sha256: hex::encode(Sha256::digest(bytes)), + size: bytes.len() as u64, + } + } + #[tokio::test] + async fn signed_index_reopens_without_whole_scan_and_seek_slices_reject_mutation() { + let root = tempfile::tempdir().unwrap(); + let identity = NodeIdentity::load_or_create(&root.path().join("identity")) + .await + .unwrap(); + let bytes: Vec = (0..CHUNK_BYTES * 5 + 19).map(|n| (n % 251) as u8).collect(); + let media = root.path().join("media"); + std::fs::write(&media, &bytes).unwrap(); + let mut file = File::open(&media).unwrap(); + let mut progress = Vec::new(); + let original = binding(&bytes); + let index = Index::scan(&mut file, original.clone(), |n| { + progress.push(n); + Ok(()) + }) + .unwrap(); + assert_eq!(progress.last(), Some(&(bytes.len() as u64))); + index.save(&root.path().join("index"), &identity).unwrap(); + drop(index); + let loaded = Index::load(&root.path().join("index"), &original, &identity) + .unwrap() + .unwrap(); + for start in [0, CHUNK_BYTES + 7, CHUNK_BYTES * 4, bytes.len() - 1] { + let got = loaded.read_slice(&mut file, start as u64, 37).unwrap(); + assert_eq!(got, bytes[start..start + got.len()]); + } + let mut corrupt = bytes.clone(); + corrupt[CHUNK_BYTES + 9] ^= 1; + std::fs::write(&media, corrupt).unwrap(); + assert!(loaded + .read_slice(&mut file, (CHUNK_BYTES + 7) as u64, 1) + .is_err()); + assert_eq!(loaded.read_slice(&mut file, 0, 5).unwrap(), &bytes[..5]); + let mut saved = std::fs::read(root.path().join("index")).unwrap(); + *saved.last_mut().unwrap() ^= 1; + std::fs::write(root.path().join("index"), saved).unwrap(); + assert!(Index::load(&root.path().join("index"), &original, &identity).is_err()); + } + #[test] + fn canceled_or_wrong_hash_scan_never_produces_index() { + let root = tempfile::tempdir().unwrap(); + let bytes = vec![7; CHUNK_BYTES * 2]; + let media = root.path().join("media"); + std::fs::write(&media, &bytes).unwrap(); + let mut file = File::open(media).unwrap(); + assert!(Index::scan(&mut file, binding(&bytes), |n| { + anyhow::ensure!(n == 0, "canceled"); + Ok(()) + }) + .is_err()); + let mut wrong = binding(&bytes); + wrong.full_sha256 = "cd".repeat(32); + assert!(Index::scan(&mut file, wrong, |_| Ok(())).is_err()); + } +} diff --git a/core/archipelago/src/rental_readiness.rs b/core/archipelago/src/rental_readiness.rs new file mode 100644 index 00000000..50bf826d --- /dev/null +++ b/core/archipelago/src/rental_readiness.rs @@ -0,0 +1,315 @@ +//! Bounded background verification. Preparing/ready never creates a lease. +use crate::rental_chunk_index::{Binding, Index}; +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, Mutex, OnceLock, + }, + time::Instant, +}; +use tokio::sync::Semaphore; +const MAX_JOBS: usize = 8; +const MAX_INDEX_MEMORY: u64 = 64 * 1024 * 1024; +#[derive(Serialize, Deserialize, Debug)] +#[serde(tag = "state", rename_all = "snake_case")] +pub(crate) enum Status { + Preparing { + completed_bytes: u64, + total_bytes: u64, + }, + Ready { + ready_id: String, + total_bytes: u64, + }, + Unavailable, +} +struct Reservation { + used: Arc, + bytes: u64, +} +impl Drop for Reservation { + fn drop(&mut self) { + self.used.fetch_sub(self.bytes, Ordering::SeqCst); + } +} +pub(crate) struct Ready { + pub index: Index, + pub id: String, + _memory: Reservation, +} +#[cfg(test)] +impl Ready { + pub(crate) fn fixture(index: Index) -> Arc { + Arc::new(Self { + index, + id: uuid::Uuid::new_v4().to_string(), + _memory: Reservation { + used: Arc::new(AtomicU64::new(0)), + bytes: 0, + }, + }) + } +} +enum State { + Preparing, + Ready(Arc), + Failed, +} +struct Job { + binding: Binding, + state: Mutex, + progress: Arc, + touched: Instant, +} +pub(crate) struct Manager { + jobs: Mutex>>, + workers: Arc, + memory: Arc, +} +impl Default for Manager { + fn default() -> Self { + Self { + jobs: Mutex::new(HashMap::new()), + workers: Arc::new(Semaphore::new(2)), + memory: Arc::new(AtomicU64::new(0)), + } + } +} +pub(crate) fn shared() -> &'static Manager { + static INSTANCE: OnceLock = OnceLock::new(); + INSTANCE.get_or_init(Manager::default) +} +impl Manager { + #[cfg(test)] + pub(crate) fn forget_for_restart(&self, key: &str) { + self.jobs.lock().unwrap().remove(key); + } + + fn status(job: &Job) -> Result { + Ok( + match &*job + .state + .lock() + .map_err(|_| anyhow::anyhow!("Media readiness unavailable"))? + { + State::Preparing => Status::Preparing { + completed_bytes: job.progress.load(Ordering::SeqCst), + total_bytes: job.binding.size, + }, + State::Ready(ready) => Status::Ready { + ready_id: ready.id.clone(), + total_bytes: job.binding.size, + }, + State::Failed => Status::Unavailable, + }, + ) + } + /// Work closure opens only the already-authorized immutable snapshot. At most + /// two closures execute, eight jobs exist, and64MiB is reserved for indexes. + /// Stream-held Arcs retain their reservation even after cache eviction. + pub fn prepare( + &self, + key: String, + binding: Binding, + work: impl FnOnce(Arc) -> Result + Send + 'static, + ) -> Result { + let mut jobs = self + .jobs + .lock() + .map_err(|_| anyhow::anyhow!("Media readiness unavailable"))?; + if let Some(job) = jobs.get(&key) { + anyhow::ensure!(job.binding == binding, "Readiness immutable terms changed"); + return Self::status(job); + } + let bytes = binding + .index_bytes()? + .checked_mul(2) + .and_then(|v| v.checked_add(128 * 1024)) + .context("Index budget overflow")?; + loop { + let used = self.memory.load(Ordering::SeqCst); + if jobs.len() < MAX_JOBS + && used + .checked_add(bytes) + .is_some_and(|n| n <= MAX_INDEX_MEMORY) + { + break; + } + let victim = jobs + .iter() + .filter(|(_, job)| { + job.state + .lock() + .map(|state| !matches!(*state, State::Preparing)) + .unwrap_or(false) + }) + .min_by_key(|(_, job)| job.touched) + .map(|(key, _)| key.clone()); + if let Some(victim) = victim { + jobs.remove(&victim); + } else { + return Ok(Status::Preparing { + completed_bytes: 0, + total_bytes: binding.size, + }); + } + } + let permit = match self.workers.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + return Ok(Status::Preparing { + completed_bytes: 0, + total_bytes: binding.size, + }) + } + }; + // Admissions are serialized by jobs; concurrent drops can only lower use. + self.memory.fetch_add(bytes, Ordering::SeqCst); + let reservation = Reservation { + used: self.memory.clone(), + bytes, + }; + let progress = Arc::new(AtomicU64::new(0)); + let job = Arc::new(Job { + binding: binding.clone(), + state: Mutex::new(State::Preparing), + progress: progress.clone(), + touched: Instant::now(), + }); + jobs.insert(key, job.clone()); + tokio::task::spawn_blocking(move || { + let _permit = permit; + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| work(progress))) + .unwrap_or_else(|_| Err(anyhow::anyhow!("Media verification worker failed"))); + let mut state = match job.state.lock() { + Ok(state) => state, + Err(_) => return, + }; + *state = match result { + Ok(index) if index.binding == binding => { + job.progress.store(binding.size, Ordering::SeqCst); + State::Ready(Arc::new(Ready { + index, + id: uuid::Uuid::new_v4().to_string(), + _memory: reservation, + })) + } + _ => State::Failed, + }; + }); + Ok(Status::Preparing { + completed_bytes: 0, + total_bytes: binding.size, + }) + } + /// Explicit retry clears only a failed preparation, never a running job or + /// a lease. The caller has already reauthenticated the original purchase. + pub fn retry_failed(&self, key: &str, binding: &Binding) -> Result<()> { + let mut jobs = self + .jobs + .lock() + .map_err(|_| anyhow::anyhow!("Media readiness unavailable"))?; + if let Some(job) = jobs.get(key) { + anyhow::ensure!(&job.binding == binding, "Readiness immutable terms changed"); + let failed = matches!( + *job.state + .lock() + .map_err(|_| anyhow::anyhow!("Media readiness unavailable"))?, + State::Failed + ); + if failed { + jobs.remove(key); + } + } + Ok(()) + } + pub fn ready( + &self, + key: &str, + binding: &Binding, + ready_id: Option<&str>, + ) -> Result> { + let jobs = self + .jobs + .lock() + .map_err(|_| anyhow::anyhow!("Media readiness unavailable"))?; + let job = jobs + .get(key) + .context("Media is preparing; no rental was started")?; + anyhow::ensure!(&job.binding == binding, "Readiness immutable terms changed"); + let state = job + .state + .lock() + .map_err(|_| anyhow::anyhow!("Media readiness unavailable"))?; + let State::Ready(ready) = &*state else { + anyhow::bail!("Media is not ready; no rental was started"); + }; + anyhow::ensure!( + ready_id.is_none_or(|id| id == ready.id), + "Media readiness changed; prepare again without paying" + ); + Ok(ready.clone()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use sha2::{Digest, Sha256}; + #[tokio::test] + async fn slow_verification_is_nonblocking_deduplicated_and_has_no_start_side_effect() { + let manager = Manager::default(); + let root = tempfile::tempdir().unwrap(); + let media = root.path().join("media"); + let bytes = vec![9; 1024 * 1024]; + std::fs::write(&media, &bytes).unwrap(); + let binding = Binding { + content_id: format!("registered_{}", uuid::Uuid::new_v4()), + receipt_sha256: "ab".repeat(32), + full_sha256: hex::encode(Sha256::digest(&bytes)), + size: bytes.len() as u64, + }; + let (release, wait) = std::sync::mpsc::channel(); + let clone = binding.clone(); + assert!(matches!( + manager + .prepare("key".into(), binding.clone(), move |progress| { + wait.recv().unwrap(); + let mut file = std::fs::File::open(media)?; + Index::scan(&mut file, clone, |n| { + progress.store(n, Ordering::SeqCst); + Ok(()) + }) + }) + .unwrap(), + Status::Preparing { .. } + )); + assert!(manager.ready("key", &binding, None).is_err()); + assert!(matches!( + manager + .prepare("key".into(), binding.clone(), |_| panic!( + "Duplicate full scan" + )) + .unwrap(), + Status::Preparing { .. } + )); + release.send(()).unwrap(); + let ready = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let Ok(ready) = manager.ready("key", &binding, None) { + break ready; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(manager.ready("key", &binding, Some("other-ready")).is_err()); + assert!(manager.ready("key", &binding, Some(&ready.id)).is_ok()); + let restarted = Manager::default(); + assert!(restarted.ready("key", &binding, Some(&ready.id)).is_err()); + } +} diff --git a/core/archipelago/src/wallet/payment_tests.rs b/core/archipelago/src/wallet/payment_tests.rs index b678c953..de7d3118 100644 --- a/core/archipelago/src/wallet/payment_tests.rs +++ b/core/archipelago/src/wallet/payment_tests.rs @@ -2214,6 +2214,10 @@ impl crate::content_purchase_caller::PurchaseTransport for PurchaseTestTransport fn seller_onion(&self) -> &str { "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.onion" } + async fn prepare_offer(&self, content_id: &str) -> anyhow::Result> { + // A registered movie still being hashed, without asking the fake mint. + Ok(content_id.starts_with("registered_").then_some((3, 16))) + } async fn offer( &self, id: &str, @@ -2339,6 +2343,36 @@ async fn full_caller_recovers_lost_acceptance_and_settlement_without_another_pay lose_settle: AtomicBool::new(true), offers: Default::default(), }; + let preparing_id = format!("registered_{}", uuid::Uuid::new_v4()); + for _ in 0..3 { + assert!(matches!( + purchase( + buyer.path(), + &buyer_did, + &preparing_id, + None, + 8, + None, + &transport + ) + .await + .unwrap(), + ReadyPurchase::Preparing { + completed_bytes: 3, + total_bytes: 16 + } + )); + } + assert!(transport.offers.lock().unwrap().is_empty()); + assert!(mint.requests.lock().unwrap().is_empty()); + assert_eq!(load_wallet(buyer.path()).await.unwrap().balance(), 8); + assert!(crate::content_purchase::Journal::open(buyer.path()) + .await + .unwrap() + .find_buyers(&buyer_did, &transport.template.seller_did, &preparing_id) + .await + .unwrap() + .is_empty()); assert!(purchase( buyer.path(), &buyer_did, diff --git a/docs/paid-content-recovery-followup.md b/docs/paid-content-recovery-followup.md index 99dde93f..0cf7b68f 100644 --- a/docs/paid-content-recovery-followup.md +++ b/docs/paid-content-recovery-followup.md @@ -772,3 +772,44 @@ The separate readiness/index work must remove those scans from request paths, verify chunks and start the original clock only after explicit ready/start. IndeeHub private app packaging, distributed announcement delivery and actual registration/payment/playback acceptance remain open. + +### Separate rental readiness candidate — not yet qualified or deployed + +The `work/rental-readiness` checkout builds on `49703d7e`; it does not change the +qualified ordinary-purchase deployment. Large-film rental activation remains +blocked by the documented full-file hashing and first-lease timeout problem. + +The draft separates bounded background verification from explicit Start. A +completed scan must match the original producer receipt's full SHA before the +node signs a 64 KiB chunk index. Signed indexes bind the receipt, full hash, size, +content identity and chunk hashes. At most two verification jobs run; index +memory and cached jobs are bounded. Preparation can finish after a disconnected +request but cannot start a lease. New registered purchase requests return +preparation progress before allocating an offer UUID or spending funds. + +Authenticated Start commits the original viewing window once. A lost response +is recovered using the same purchase, even if its ephemeral readiness ID was +lost. Metadata and range reads do not initiate whole-file verification or leases. +Each returned range slice is taken from a complete chunk whose hash was checked +immediately before delivery. A changed chunk stops delivery; it does not renew +the window or authorize another payment. Existing signed indexes avoid a full +scan after restart, but do not establish that every current media byte is still +present: subsequent corruption remains a recoverable delivery failure. + +The added tests cover preparation without a buyer intent or mint request, +settlement/preparation without a lease, GET refusal before Start, original-window +replay, repeated range reads with a full-scan counter, signature/index alteration +and changed chunk rejection. These changes have only passed formatting and diff +checks so far. Isolated compilation, regression tests, actual large-file timing, +installed-app consent/Start flows and recovery acceptance are still outstanding. + +The matching host draft advertises read-only `archipelagoRental.playbackProtocol` +2 and requires `request(offer, {playbackProtocol: 2, signal?})`. Paid replies expose +only protocol, opaque handle, operation ID and known expiry. Separate installed- +frame `prepare`, `start` and `status` actions revalidate the native app context. +Only a successful explicit Start returns a playback URL. Preparation progress is +visible in overlay, app-session and tab-signer consent surfaces. Polling is capped; +closing, aborting or timing out cancels only its own pending prompt. A payment +already dispatched remains journaled and is recovered using its original ID. +The host rejects malformed states, changed observed windows and playback URLs in +non-started replies. The focused host/provider run passed 29 tests across two files; actual `vue-tsc -b` passed, with all 542 captured host inputs unchanged. Logs are `/tmp/archy-rental-protocol2-host-tests.log` and `/tmp/archy-rental-protocol2-host-typecheck.log`; provenance is `/tmp/archy-rental-protocol2-host-provenance.json`. The rental Rust remains uncompiled pending the combined backend candidate. diff --git a/neode-ui/public/nostr-provider.js b/neode-ui/public/nostr-provider.js index 9f6deea4..fe2c9972 100644 --- a/neode-ui/public/nostr-provider.js +++ b/neode-ui/public/nostr-provider.js @@ -291,7 +291,7 @@ if (e.data.type === 'archipelago-rental-response') { var rental = rentalPending[e.data.id]; if (!rental) return; - delete rentalPending[e.data.id]; clearTimeout(rental.timer); + delete rentalPending[e.data.id]; clearTimeout(rental.timer); rental.cleanup(); e.data.error ? rental.reject(new Error(e.data.error)) : rental.resolve(e.data.result); return; } @@ -336,26 +336,36 @@ return hex.slice(0,8)+'-'+hex.slice(8,12)+'-'+hex.slice(12,16)+'-'+hex.slice(16,20)+'-'+hex.slice(20); } + function rentalRequest(action, payload, timeout, signal) { + return new Promise(function (resolve, reject) { + if (signal && (typeof signal.addEventListener !== 'function' || typeof signal.removeEventListener !== 'function' || typeof signal.aborted !== 'boolean')) { reject(new Error('Invalid rental cancellation signal.')); return; } + if (signal && signal.aborted) { reject(new Error('Rental request closed.')); return; } + var id = nativeRequestId(); + function cleanup() { if (signal) signal.removeEventListener('abort', abort); } + function fail(message) { + var pending = rentalPending[id]; if (!pending) return; + delete rentalPending[id]; clearTimeout(pending.timer); cleanup(); + if (action === 'request') postToSigner({type:'archipelago-rental-request',id:id,action:'cancel',playbackProtocol:2}); + reject(new Error(message)); + } + function abort() { fail('Rental request closed. Recover any dispatched payment using the original purchase.'); } + rentalPending[id] = { resolve: resolve, reject: reject, cleanup: cleanup, timer: setTimeout(function () { + fail('Rental response unavailable. Recover the same purchase without paying again.'); + }, timeout) }; + if (signal) signal.addEventListener('abort', abort, {once:true}); + postToSigner(Object.assign({ type: 'archipelago-rental-request', id: id, action: action, playbackProtocol: 2 }, payload)); + }); + } window.archipelagoRental = { - status: function (handle) { - return new Promise(function (resolve, reject) { - var id = nativeRequestId(); - rentalPending[id] = { resolve: resolve, reject: reject, timer: setTimeout(function () { - delete rentalPending[id]; reject(new Error('Playback status unavailable.')); - }, 15000) }; - postToSigner({ type: 'archipelago-rental-request', id: id, action: 'status', handle: handle }); - }); - }, - request: function (offer) { - return new Promise(function (resolve, reject) { - var id = nativeRequestId(); - rentalPending[id] = { resolve: resolve, reject: reject, timer: setTimeout(function () { - delete rentalPending[id]; reject(new Error('Rental response unavailable. Reopen this title to recover the same purchase.')); - }, 600000) }; - postToSigner({ type: 'archipelago-rental-request', id: id, offer: offer }); - }); + status: function (handle) { return rentalRequest('status', { handle: handle }, 35000); }, + prepare: function (handle, retry) { return rentalRequest('prepare', { handle: handle, retry: retry === true }, 35000); }, + start: function (handle, readyId) { return rentalRequest('start', { handle: handle, ready_id: readyId }, 35000); }, + request: function (offer, options) { + if (!options || options.playbackProtocol !== 2) return Promise.reject(new Error('Playback protocol 2 is required before requesting a rental.')); + return rentalRequest('request', { offer: offer }, 600000, options.signal); } }; + Object.defineProperty(window.archipelagoRental, 'playbackProtocol', { value: 2, writable: false, configurable: false, enumerable: true }); window.archipelagoMediaRegistration = { request: function (action, payload) { diff --git a/neode-ui/src/components/AppLauncherOverlay.vue b/neode-ui/src/components/AppLauncherOverlay.vue index dfd1e15c..302e987e 100644 --- a/neode-ui/src/components/AppLauncherOverlay.vue +++ b/neode-ui/src/components/AppLauncherOverlay.vue @@ -181,7 +181,7 @@ - + diff --git a/neode-ui/src/components/RentalPurchaseConsent.vue b/neode-ui/src/components/RentalPurchaseConsent.vue index 06a1feeb..0a31cb0b 100644 --- a/neode-ui/src/components/RentalPurchaseConsent.vue +++ b/neode-ui/src/components/RentalPurchaseConsent.vue @@ -3,10 +3,10 @@ @@ -16,10 +16,10 @@ import { computed, ref } from 'vue' import { useModalKeyboard } from '@/composables/useModalKeyboard' import type { RentalOffer } from '@/composables/useRentalPurchaseBridge' -const props = defineProps<{ request: RentalOffer | null; quote: { wallet_debit_sats: number; network: 'mainnet' | 'testnet'; mint_url: string } | null; phase: string; error: string }>() +const props = defineProps<{ request: RentalOffer | null; quote: { wallet_debit_sats: number; network: 'mainnet' | 'testnet'; mint_url: string } | null; phase: string; error: string; progress?: string }>() const emit = defineEmits<{ review: []; approve: []; cancel: []; cancelUnpaid: [] }>() const modal = ref(null) useModalKeyboard(modal, computed(() => Boolean(props.request)), () => emit('cancel')) -const busy = computed(() => props.phase === 'loading' || props.phase === 'paying') +const busy = computed(() => props.phase === 'preparing' || props.phase === 'loading' || props.phase === 'paying') diff --git a/neode-ui/src/composables/__tests__/nativeProviderRequests.test.ts b/neode-ui/src/composables/__tests__/nativeProviderRequests.test.ts index c00d2ab6..bc8f6114 100644 --- a/neode-ui/src/composables/__tests__/nativeProviderRequests.test.ts +++ b/neode-ui/src/composables/__tests__/nativeProviderRequests.test.ts @@ -10,7 +10,11 @@ describe('native provider on ordinary HTTP node origins',()=>{ const timers=new Set();let count=0 runInNewContext(source,{window,document:{currentScript:{hasAttribute:()=>true},readyState:'complete'},crypto:{getRandomValues:(bytes:Uint8Array)=>{bytes.fill(++count);return bytes}},Uint8Array,URL,console, setTimeout:(fn:unknown)=>{timers.add(fn);return fn},clearTimeout:(fn:unknown)=>timers.delete(fn)}) - const rental=window.archipelagoRental.request({title:'Film'}) + expect(window.archipelagoRental.playbackProtocol).toBe(2) + expect(Object.getOwnPropertyDescriptor(window.archipelagoRental,'playbackProtocol')?.writable).toBe(false) + await expect(window.archipelagoRental.request({title:'Legacy'})).rejects.toThrow('protocol 2') + expect(parent.postMessage).not.toHaveBeenCalled() + const rental=window.archipelagoRental.request({title:'Film'},{playbackProtocol:2}) const registration=window.archipelagoMediaRegistration.request('resume',{intent:{requestId:'existing'}}) const requests=parent.postMessage.mock.calls.map(([message])=>message) expect(requests).toHaveLength(2) @@ -19,5 +23,19 @@ describe('native provider on ordinary HTTP node origins',()=>{ for(const message of requests) for(const receive of listeners) receive({source:parent,origin:'http://node.local',data:{type:message.type.replace('-request','-response'),id:message.id,result:{ok:true}}}) await expect(rental).resolves.toEqual({ok:true});await expect(registration).resolves.toEqual({ok:true}) expect(timers.size).toBe(0) + parent.postMessage.mockClear() + const stop = new AbortController() + const closed = window.archipelagoRental.request({title:'Closed'}, {playbackProtocol:2,signal:stop.signal}) + const closedCheck = expect(closed).rejects.toThrow('Rental request closed') + const retained = window.archipelagoRental.request({title:'Other'}, {playbackProtocol:2}) + const messages=parent.postMessage.mock.calls.map(([message])=>message) + stop.abort();await closedCheck + expect(parent.postMessage.mock.lastCall![0]).toMatchObject({action:'cancel',id:messages[0].id,playbackProtocol:2}) + expect(timers.size).toBe(1) + for(const receive of listeners)receive({source:parent,origin:'http://node.local',data:{type:'archipelago-rental-response',id:messages[1].id,result:{playbackProtocol:2,handle:'d'.repeat(64)}}}) + await expect(retained).resolves.toMatchObject({playbackProtocol:2});expect(timers.size).toBe(0) + parent.postMessage.mockClear() + await expect(window.archipelagoRental.request({title:'Already closed'},{playbackProtocol:2,signal:stop.signal})).rejects.toThrow('closed') + expect(parent.postMessage).not.toHaveBeenCalled() }) }) diff --git a/neode-ui/src/composables/__tests__/useRentalPurchaseBridge.test.ts b/neode-ui/src/composables/__tests__/useRentalPurchaseBridge.test.ts index e6bbbef8..5f640999 100644 --- a/neode-ui/src/composables/__tests__/useRentalPurchaseBridge.test.ts +++ b/neode-ui/src/composables/__tests__/useRentalPurchaseBridge.test.ts @@ -8,11 +8,11 @@ const quote = { state: 'confirmation_required', network: 'mainnet', mint_url: 'h function fixture(consentBusy: () => boolean = () => false) { const child = { postMessage: vi.fn() } const bridge = useRentalPurchaseBridge({ consentBusy, appId: () => 'indeedhub', appUrl: () => 'https://node.test:7778/browse', frameWindow: () => child as unknown as Window }) - const send = (origin = 'https://node.test:7778', source: unknown = child) => bridge.handle({ data: { type: 'archipelago-rental-request', id: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', offer }, origin, source } as MessageEvent) + const send = (origin = 'https://node.test:7778', source: unknown = child) => bridge.handle({ data: { type: 'archipelago-rental-request', playbackProtocol: 2, id: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', offer }, origin, source } as MessageEvent) return { bridge, child, send } } -afterEach(() => vi.unstubAllGlobals()) -beforeEach(() => { vi.stubGlobal('location', new URL('https://node.test')); vi.clearAllMocks(); rpc.call.mockImplementation(async ({ method }) => method === 'media.registration.context' ? installed : method === 'content.rental-purchase' ? quote : { playback_url: '/api/rental-playback/' + 'd'.repeat(64), expires_at: null }) }) +afterEach(() => { vi.unstubAllGlobals(); vi.useRealTimers() }) +beforeEach(() => { vi.stubGlobal('location', new URL('https://node.test')); vi.clearAllMocks(); rpc.call.mockImplementation(async ({ method }) => method === 'media.registration.context' ? installed : method === 'content.rental-purchase' ? quote : { handle: 'd'.repeat(64), expires_at: null }) }) describe('native rental confirmation', () => { it('cannot approve a quote without its saved network and mint', async()=>{ const f=fixture();await f.send() @@ -26,12 +26,12 @@ describe('native rental confirmation', () => { const f=fixture(); await f.send(); await f.bridge.approve(); expect(rpc.call).toHaveBeenCalledTimes(1) await f.bridge.review(); expect(f.bridge.phase.value).toBe('confirm') expect(rpc.call.mock.calls.find(([v]) => v.method==='content.rental-purchase')![0].params.consent).toBeUndefined() - rpc.call.mockImplementation(async ({method})=>method==='media.registration.context'?installed:method==='content.rental-purchase'?{state:'entitled',operation_id:quote.operation_id}:{playback_url:'/api/rental-playback/'+'d'.repeat(64),expires_at:null}) + rpc.call.mockImplementation(async ({method})=>method==='media.registration.context'?installed:method==='content.rental-purchase'?{state:'entitled',operation_id:quote.operation_id}:{handle:'d'.repeat(64),expires_at:null}) await f.bridge.approve() const calls=rpc.call.mock.calls.filter(([v])=>v.method==='content.rental-purchase') expect(calls[1]![0].params.consent).toEqual({operation_id:quote.operation_id,envelope_sha256:quote.envelope_sha256,wallet_debit_sats:10}) expect(calls[1]![0].params.max_wallet_debit).toBe(10); expect(f.bridge.request.value).toBeNull() - expect(f.child.postMessage.mock.lastCall![0].result.playback_url).toContain('/api/rental-playback/') + expect(f.child.postMessage.mock.lastCall![0].result).toEqual({playbackProtocol:2,handle:'d'.repeat(64),expires_at:null,operation_id:quote.operation_id}); expect(rpc.call.mock.calls.some(([v])=>v.method==='content.playback-start')).toBe(false) }) it('does not dispatch after closing during installation validation', async()=>{ const f=fixture(); await f.send(); let release!:(value:unknown)=>void @@ -63,7 +63,7 @@ describe('native rental confirmation', () => { }) it('lease status does not open a purchase or confirm spending', async()=>{ const f=fixture();rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:{expires_at:null}) - await f.bridge.handle({data:{type:'archipelago-rental-request',id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'status',handle:'d'.repeat(64)},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + await f.bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'status',handle:'d'.repeat(64)},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) expect(rpc.call.mock.calls.map(([v])=>v.method)).toEqual(['media.registration.context','content.playback-status']) expect(f.child.postMessage.mock.lastCall![0].result).toEqual({expires_at:null}) expect(f.bridge.request.value).toBeNull() @@ -83,7 +83,7 @@ describe('native rental confirmation', () => { it('drops a status response after the surface closes even if its WindowProxy is reused',async()=>{ const f=fixture();let release!:(value:unknown)=>void rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:new Promise(resolve=>{release=resolve})) - const work=f.bridge.handle({data:{type:'archipelago-rental-request',id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'status',handle:'d'.repeat(64)},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + const work=f.bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'status',handle:'d'.repeat(64)},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) await vi.waitFor(()=>expect(release).toBeTypeOf('function')) f.bridge.cancel();release({expires_at:100});await work expect(f.child.postMessage).not.toHaveBeenCalled() @@ -91,7 +91,7 @@ describe('native rental confirmation', () => { it('does not dispatch when the active frame disappears during installation verification',async()=>{ const child={postMessage:vi.fn()};let active=true const bridge=useRentalPurchaseBridge({appId:()=> 'indeedhub',appUrl:()=> 'https://node.test:7778/browse',frameWindow:()=>active?child as unknown as Window:null}) - await bridge.handle({data:{type:'archipelago-rental-request',id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',offer},origin:'https://node.test:7778',source:child} as unknown as MessageEvent) + await bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',offer},origin:'https://node.test:7778',source:child} as unknown as MessageEvent) let release!:(value:unknown)=>void;rpc.call.mockImplementationOnce(()=>new Promise(resolve=>{release=resolve})) const work=bridge.review();active=false;bridge.cancel();release(installed);await work expect(rpc.call.mock.calls.some(([v])=>v.method==='content.rental-purchase')).toBe(false) @@ -111,3 +111,87 @@ describe('native rental confirmation', () => { }) }) + +describe('protocol 2 preparation and explicit Start',()=>{ + it('rejects legacy callers before a wallet request',async()=>{ + const f=fixture();await f.bridge.handle({data:{type:'archipelago-rental-request',id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',offer},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + expect(rpc.call).not.toHaveBeenCalled();expect(f.child.postMessage.mock.lastCall![0].error).toContain('protocol 2') + }) + it('closing while preparation waits cancels polling and never approves payment',async()=>{ + vi.useFakeTimers(); const f=fixture();await f.send() + rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:{state:'preparing',completed_bytes:4,total_bytes:16}) + const work=f.bridge.review();await vi.advanceTimersByTimeAsync(0) + expect(f.bridge.phase.value).toBe('preparing');expect(f.bridge.progress.value).toContain('25%') + f.bridge.cancel();await work;await vi.advanceTimersByTimeAsync(10000) + const payments=rpc.call.mock.calls.filter(([v])=>v.method==='content.rental-purchase');expect(payments).toHaveLength(1);expect(payments[0]![0].params.consent).toBeUndefined() + }) + it('preparation reaches confirmation without automatically accepting it',async()=>{ + vi.useFakeTimers();const f=fixture();await f.send();let polls=0 + rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:++polls===1?{state:'preparing',completed_bytes:0,total_bytes:16}:quote) + const work=f.bridge.review();await vi.advanceTimersByTimeAsync(2000);await work + expect(f.bridge.phase.value).toBe('confirm');expect(rpc.call.mock.calls.filter(([v])=>v.method==='content.rental-purchase').every(([v])=>v.params.consent===undefined)).toBe(true) + }) + it('only a distinct Start action returns the local playback URL',async()=>{ + const f=fixture();const ready='aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:method==='content.playback-prepare'?{state:'ready',ready_id:ready,handle:'d'.repeat(64),viewing_seconds:3600,started_at:null,expires_at:null}:{state:'started',started_at:100,expires_at:3700,playback_url:'/api/rental-playback/'+'d'.repeat(64)}) + const event=(action:string)=>({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action,handle:'d'.repeat(64),ready_id:ready},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + await f.bridge.handle(event('prepare'));expect(f.child.postMessage.mock.lastCall![0].result.playback_url).toBeUndefined();expect(rpc.call.mock.calls.some(([v])=>v.method==='content.playback-start')).toBe(false) + await f.bridge.handle(event('start'));expect(f.child.postMessage.mock.lastCall![0].result.playback_url).toBe('https://node.test/api/rental-playback/'+'d'.repeat(64)) + expect(rpc.call.mock.calls.filter(([v])=>v.method==='media.registration.context')).toHaveLength(2) + }) +}) + +describe('provider request cancellation ownership',()=>{ + it('cancels only the matching native request and cannot approve its old quote',async()=>{ + const f=fixture();await f.send();await f.bridge.review() + const event=(id:string)=>({data:{type:'archipelago-rental-request',playbackProtocol:2,action:'cancel',id},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + await f.bridge.handle(event('bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'));expect(f.bridge.phase.value).toBe('confirm') + await f.bridge.handle(event('cccccccc-cccc-4ccc-8ccc-cccccccccccc'));expect(f.bridge.request.value).toBeNull() + const calls=rpc.call.mock.calls.length;await f.bridge.approve();expect(rpc.call).toHaveBeenCalledTimes(calls) + }) + it('drops Start after closing during installed-frame verification',async()=>{ + const f=fixture();let release!:(value:unknown)=>void + rpc.call.mockImplementationOnce(()=>new Promise(resolve=>{release=resolve})) + const work=f.bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'start',handle:'d'.repeat(64),ready_id:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + f.bridge.cancel();release(installed);await work + expect(rpc.call.mock.calls.some(([v])=>v.method==='content.playback-start')).toBe(false) + }) +}) + +describe('strict protocol responses and approved-operation recovery',()=>{ + it('rejects unknown actions rather than opening a purchase',async()=>{ + const f=fixture();await f.bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'unexpected',offer},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + expect(rpc.call).not.toHaveBeenCalled();expect(f.bridge.request.value).toBeNull() + }) + it('preserves approved operation when a preparing response arrives after consent',async()=>{ + const f=fixture();await f.send();await f.bridge.review() + rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:{state:'preparing',completed_bytes:0,total_bytes:16}) + await f.bridge.approve();expect(f.bridge.error.value).toContain('Payment was approved');expect(f.bridge.error.value).toContain('original operation');expect(f.bridge.quote.value?.operation_id).toBe(quote.operation_id) + const calls=rpc.call.mock.calls.length;await f.bridge.approve();expect(rpc.call).toHaveBeenCalledTimes(calls) + }) + it.each([ + {state:'ready',ready_id:'invalid',handle:'d'.repeat(64),viewing_seconds:60,started_at:null,expires_at:null}, + {state:'preparing',completed_bytes:20,total_bytes:10,viewing_seconds:60,started_at:null,expires_at:null}, + {state:'ready',ready_id:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa',handle:'d'.repeat(64),viewing_seconds:0,started_at:null,expires_at:null}, + {state:'expired',started_at:100,expires_at:200,playback_url:'/api/rental-playback/'+'d'.repeat(64)}, + ])('rejects malformed or URL-bearing non-started states',async(result)=>{ + const f=fixture();rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:result) + await f.bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'prepare',handle:'d'.repeat(64)},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + expect(f.child.postMessage.mock.lastCall![0].error).toContain('Playback state');expect(f.child.postMessage.mock.lastCall![0].result).toBeUndefined() + }) + it('drops a completed Start response from a closed frame generation',async()=>{ + const f=fixture();let release!:(value:unknown)=>void + rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:new Promise(resolve=>{release=resolve})) + const work=f.bridge.handle({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action:'start',handle:'d'.repeat(64),ready_id:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + await vi.waitFor(()=>expect(release).toBeTypeOf('function'));f.bridge.cancel();release({state:'started',started_at:100,expires_at:200,playback_url:'/api/rental-playback/'+'d'.repeat(64)});await work + expect(f.child.postMessage).not.toHaveBeenCalled() + }) +}) + +it('rejects a changed original viewing window on a recovered Start',async()=>{ + const f=fixture();const handle='d'.repeat(64),ready='aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + rpc.call.mockImplementation(async({method})=>method==='media.registration.context'?installed:method==='content.playback-prepare'?{state:'ready',ready_id:ready,handle,viewing_seconds:3600,started_at:100,expires_at:3700}:{state:'started',started_at:200,expires_at:3800,playback_url:'/api/rental-playback/'+handle}) + const event=(action:string)=>({data:{type:'archipelago-rental-request',playbackProtocol:2,id:'cccccccc-cccc-4ccc-8ccc-cccccccccccc',action,handle,ready_id:ready},origin:'https://node.test:7778',source:f.child} as unknown as MessageEvent) + await f.bridge.handle(event('prepare'));await f.bridge.handle(event('start')) + expect(f.child.postMessage.mock.lastCall![0].error).toContain('original viewing window');expect(f.child.postMessage.mock.lastCall![0].result).toBeUndefined() +}) diff --git a/neode-ui/src/composables/useRentalPurchaseBridge.ts b/neode-ui/src/composables/useRentalPurchaseBridge.ts index c147d7ff..70f01be1 100644 --- a/neode-ui/src/composables/useRentalPurchaseBridge.ts +++ b/neode-ui/src/composables/useRentalPurchaseBridge.ts @@ -6,6 +6,47 @@ interface FrameContext { appId: () => string; appUrl: () => string; frameWindow: export interface RentalOffer { title: string; terms: { nodeDid: string; contentId: string; sha256: string; priceSats: number; viewingSeconds: number } } interface Quote { network: 'mainnet' | 'testnet'; mint_url: string; state: string; operation_id: string; envelope_sha256: string; wallet_debit_sats: number; gross_token_sats: number; seller_net_sats: number; expires_at: number; seller_onion: string } interface Pending { source: Window; origin: string; id: string; offer: RentalOffer; quote?: Quote } +type PlaybackAction = 'status' | 'prepare' | 'start' +interface PlaybackBinding { duration?: number; startedAt?: number; expiresAt?: number } +const readyPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/ +const positiveInteger = (value: unknown): value is number => Number.isSafeInteger(value) && (value as number) > 0 +function validatePlayback(raw: unknown, action: PlaybackAction, handle: string, original: PlaybackBinding): { result: Record; binding: PlaybackBinding } { + if (!raw || typeof raw !== 'object' || Array.isArray(raw)) throw new Error('Invalid playback response.') + const data = raw as Record, binding = { ...original } + const failure = () => { throw new Error('Playback state or original viewing window changed. Recover this purchase without paying again.') } + const expires = data.expires_at + if (!(expires === null || positiveInteger(expires))) return failure() + if (binding.expiresAt !== undefined && expires !== binding.expiresAt) return failure() + if (positiveInteger(expires)) binding.expiresAt = expires + if (action === 'status') { + if ('playback_url' in data) return failure() + return { result: { expires_at: expires }, binding } + } + const state = data.state + if (state !== 'started' && 'playback_url' in data) return failure() + if (state === 'unavailable' && action === 'prepare') return { result: {state, expires_at: expires}, binding } + const start = data.started_at + if (!((start === null && expires === null) || (positiveInteger(start) && positiveInteger(expires) && expires > start))) return failure() + if (binding.startedAt !== undefined && start !== binding.startedAt) return failure() + if (positiveInteger(start)) binding.startedAt = start + if (state === 'preparing' || state === 'ready') { + if (action !== 'prepare' || !positiveInteger(data.viewing_seconds)) return failure() + if (binding.duration !== undefined && data.viewing_seconds !== binding.duration) return failure() + binding.duration = data.viewing_seconds + if (positiveInteger(start) && expires !== start + binding.duration) return failure() + const base = {state, viewing_seconds: binding.duration, started_at: start, expires_at: expires} + if (state === 'preparing') { + if (!positiveInteger(data.total_bytes) || !Number.isSafeInteger(data.completed_bytes) || (data.completed_bytes as number) < 0 || (data.completed_bytes as number) > data.total_bytes) return failure() + return { result: {...base,completed_bytes:data.completed_bytes,total_bytes:data.total_bytes}, binding } + } + if (data.handle !== handle || typeof data.ready_id !== 'string' || !readyPattern.test(data.ready_id)) return failure() + return { result: {...base,handle,ready_id:data.ready_id}, binding } + } + if (!positiveInteger(start) || !positiveInteger(expires) || (binding.duration !== undefined && expires !== start + binding.duration)) return failure() + if (state === 'expired') return { result: {state,started_at:start,expires_at:expires}, binding } + if (state !== 'started' || action !== 'start' || data.playback_url !== `/api/rental-playback/${handle}`) return failure() + return { result: {state,started_at:start,expires_at:expires,playback_url:new URL(data.playback_url as string,window.location.origin).href}, binding } +} export function supportsRentalPlaybackOrigin(appOrigin: string, dashboardOrigin: string): boolean { try { const app = new URL(appOrigin), dashboard = new URL(dashboardOrigin) @@ -17,7 +58,10 @@ export function supportsRentalPlaybackOrigin(appOrigin: string, dashboardOrigin: } export function useRentalPurchaseBridge(context: FrameContext) { const request = shallowRef(null), quote = shallowRef(null) - const phase = ref<'review' | 'loading' | 'confirm' | 'paying'>('review'), error = ref('') + const phase = ref<'review' | 'loading' | 'preparing' | 'confirm' | 'paying'>('review'), error = ref('') + const progress = ref('') + const playbackBindings = new Map() + let stopWait: (() => void) | undefined let pending: Pending | null = null, disposed = false, generation = 0 const frameOrigin = () => { try { return new URL(context.appUrl(), window.location.origin).origin } catch { return '' } } const current = (item: Pending) => !disposed && pending === item && context.appId() === 'indeedhub' @@ -29,6 +73,7 @@ export function useRentalPurchaseBridge(context: FrameContext) { function cancel() { if (pending) reply(pending, undefined, phase.value === 'paying' ? 'Payment may be processing. Reopen this title to recover the same purchase.' : 'Rental confirmation closed. No new payment was approved.') + stopWait?.(); stopWait = undefined; progress.value = '' generation++ pending = null; request.value = null; quote.value = null; error.value = ''; phase.value = 'review' } @@ -41,30 +86,48 @@ export function useRentalPurchaseBridge(context: FrameContext) { if (!item || !current(item) || (confirm ? phase.value !== 'confirm' : phase.value !== 'review')) return if (!supportsRentalPlaybackOrigin(item.origin, window.location.origin)) { error.value = 'Open the app from the same node dashboard address before paying.'; return } if (context.consentBusy?.()) { error.value = 'Finish the other native confirmation first.'; return } - phase.value = confirm ? 'paying' : 'loading'; error.value = '' + phase.value = confirm ? 'paying' : 'loading'; error.value = ''; progress.value = '' try { + const terms = item.offer.terms + const deadline = Date.now() + 300000 + let result!: Quote & { completed_bytes?: number; total_bytes?: number } + for (let attempt = 0; attempt < 150; attempt++) { await installed(item) if (!current(item)) return if (context.consentBusy?.()) throw new Error('Finish the other native confirmation first.') - const terms = item.offer.terms - const result = await rpcClient.call({ method: 'content.rental-purchase', params: { + if (Date.now() >= deadline) throw new Error('Preparation is still running. Check purchase again; no new payment was approved.') + result = await rpcClient.call({ method: 'content.rental-purchase', params: { seller_did: terms.nodeDid, content_id: terms.contentId, expected_sha256: terms.sha256, expected_price_sats: terms.priceSats, expected_viewing_seconds: terms.viewingSeconds, + ...(!confirm && attempt === 0 ? { retry_preparation: true } : {}), max_wallet_debit: confirm ? item.quote!.wallet_debit_sats : Number.MAX_SAFE_INTEGER, ...(confirm ? { consent: { operation_id: item.quote!.operation_id, envelope_sha256: item.quote!.envelope_sha256, wallet_debit_sats: item.quote!.wallet_debit_sats } } : {}), - }, timeout: 120000 }) + }, timeout: Math.min(120000, deadline - Date.now()) }) if (!current(item)) return + if (result.state !== 'preparing') break + if (confirm) throw new Error('Payment was approved for the original operation. Check the original purchase to recover its payment and preparation; do not pay again.') + if (!Number.isSafeInteger(result.completed_bytes) || !Number.isSafeInteger(result.total_bytes) + || result.total_bytes! < 1 || result.completed_bytes! < 0 || result.completed_bytes! > result.total_bytes!) throw new Error('Invalid media preparation response. No new payment approved.') + phase.value = 'preparing' + progress.value = `Preparing video: ${Math.floor(result.completed_bytes! / result.total_bytes! * 100)}%. Your viewing time has not started.` + if (attempt === 149) throw new Error('Video preparation is still running. Check purchase again to continue; no new payment was approved.') + await new Promise(resolve => { const timer = setTimeout(() => { stopWait = undefined; resolve() }, 2000); stopWait = () => { clearTimeout(timer); resolve() } }) + if (!current(item)) return + } + progress.value = '' if (result.state === 'confirmation_required') { if (!validCashuIdentity(result) || !Number.isSafeInteger(result.wallet_debit_sats) || result.wallet_debit_sats < terms.priceSats || !/^[0-9a-f]{64}$/.test(result.envelope_sha256) || !result.operation_id) throw new Error('Invalid payment confirmation. No payment approved.') item.quote = result; quote.value = result; phase.value = 'confirm'; return } if (result.state !== 'entitled') throw new Error(result.state === 'cancelled_unspent' ? 'This purchase was cancelled without spending.' : 'Purchase has not completed. Reopen this title to recover it.') - const playback = await rpcClient.call<{ playback_url: string; expires_at: number | null }>({ method: 'content.playback-handle', params: { purchase_id: result.operation_id } }) + if (!readyPattern.test(result.operation_id)) throw new Error('Invalid original purchase identifier. Recover the original payment.') + const playback = await rpcClient.call<{ handle: string; expires_at: number | null }>({ method: 'content.playback-handle', params: { purchase_id: result.operation_id } }) if (!current(item)) return - if (!/^\/api\/rental-playback\/[0-9a-f]{64}$/.test(playback.playback_url)) throw new Error('Invalid playback address.') - reply(item, { ...playback, playback_url: new URL(playback.playback_url, window.location.origin).href, operation_id: result.operation_id }) + if (!/^[0-9a-f]{64}$/.test(playback.handle) || !(playback.expires_at === null || positiveInteger(playback.expires_at))) throw new Error('Invalid playback handle.') + playbackBindings.set(`${item.origin}:${playback.handle}`, {duration:terms.viewingSeconds,...(playback.expires_at === null ? {} : {expiresAt:playback.expires_at})}) + reply(item, { playbackProtocol: 2, handle: playback.handle, expires_at: playback.expires_at, operation_id: result.operation_id }) pending = null; request.value = null; quote.value = null } catch (cause) { if (current(item)) { error.value = cause instanceof Error ? cause.message : 'Could not complete this purchase. Retry to recover its original result.'; phase.value = 'review' } @@ -72,7 +135,7 @@ export function useRentalPurchaseBridge(context: FrameContext) { } async function cancelUnpaid() { const item = pending - if (!item?.quote || !current(item) || phase.value === 'paying' || phase.value === 'loading') return + if (!item?.quote || !current(item) || phase.value === 'paying' || phase.value === 'loading' || phase.value === 'preparing') return phase.value = 'loading'; error.value = '' try { await installed(item) @@ -90,21 +153,34 @@ export function useRentalPurchaseBridge(context: FrameContext) { || event.origin !== frameOrigin() || event.data?.type !== 'archipelago-rental-request') return const { id, offer } = event.data if (typeof id !== 'string' || !/^[0-9a-f-]{36}$/.test(id)) return - if (event.data.action === 'status') { + if (event.data.action === 'cancel') { if (pending?.id === id && event.data.playbackProtocol === 2) cancel(); return } + if (['status', 'prepare', 'start'].includes(event.data.action)) { const source = event.source as Window, origin = event.origin, handle = event.data.handle, scope = generation + const action = event.data.action as PlaybackAction, readyId = event.data.ready_id + if (event.data.playbackProtocol !== 2) { source.postMessage({type:'archipelago-rental-response',id,error:'Playback protocol 2 is required.'},origin); return } + if (action === 'start' && !supportsRentalPlaybackOrigin(origin, window.location.origin)) { source.postMessage({type:'archipelago-rental-response',id,error:'Open the app from this node dashboard before starting playback.'},origin); return } + if (action === 'start' && (typeof readyId !== 'string' || !/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(readyId))) return + if (action === 'prepare' && event.data.retry !== undefined && typeof event.data.retry !== 'boolean') return const selected = () => !disposed && generation === scope && context.appId() === 'indeedhub' && source === context.frameWindow() && frameOrigin() === origin if (typeof handle !== 'string' || !/^[0-9a-f]{64}$/.test(handle)) return try { const value = await rpcClient.call<{ appId: string; appOrigins: string[] }>({ method: 'media.registration.context', params: {} }) if (!selected()) return if (value.appId !== 'indeedhub' || !value.appOrigins.some(expected => installedOriginMatches(origin, expected))) throw new Error('Installed app changed.') - const result = await rpcClient.call<{ expires_at: number | null }>({ method: 'content.playback-status', params: { handle } }) - if (selected()) source.postMessage({ type: 'archipelago-rental-response', id, result }, origin) + const result = await rpcClient.call>({ method: `content.playback-${action}`, params: { handle, + ...(action === 'prepare' ? { retry: event.data.retry === true } : {}), ...(action === 'start' ? {ready_id: readyId} : {}) }, timeout: 30000 }) + if (!selected()) return + const key = `${origin}:${handle}` + const validated = validatePlayback(result, action, handle, playbackBindings.get(key) ?? {}) + playbackBindings.set(key, validated.binding) + source.postMessage({ type: 'archipelago-rental-response', id, result: validated.result }, origin) } catch (cause) { if (selected()) source.postMessage({ type: 'archipelago-rental-response', id, error: String(cause) }, origin) } return } + if (event.data.action !== undefined && event.data.action !== 'request') { (event.source as Window).postMessage({type:'archipelago-rental-response',id,error:'Unsupported rental action.'},event.origin); return } + if (event.data.playbackProtocol !== 2) { (event.source as Window).postMessage({type:'archipelago-rental-response',id,error:'Update this app to playback protocol 2 before renting. No payment was requested.'},event.origin); return } if (!supportsRentalPlaybackOrigin(event.origin, window.location.origin)) { (event.source as Window).postMessage({ type: 'archipelago-rental-response', id, error: 'Open IndeeHub from this node’s dashboard using the same LAN hostname and HTTP/HTTPS scheme before renting. This app address cannot receive the playback session cookie; no payment was requested.' }, event.origin); return } if (context.consentBusy?.()) { (event.source as Window).postMessage({ type: 'archipelago-rental-response', id, error: 'Finish the other native confirmation first.' }, event.origin); return } if (pending) { (event.source as Window).postMessage({ type: 'archipelago-rental-response', id, error: 'Finish or close the current rental confirmation first.' }, event.origin); return } @@ -121,6 +197,6 @@ export function useRentalPurchaseBridge(context: FrameContext) { try { await installed(item); if (current(item)) { request.value = item.offer; phase.value = 'review' } } catch (cause) { reply(item, undefined, String(cause)); if (pending === item) pending = null } } - return { isBusy: () => pending !== null, request, quote, phase, error, handle, cancelUnpaid, review: () => run(false), approve: () => run(true), cancel, + return { isBusy: () => pending !== null, request, quote, phase, error, progress, handle, cancelUnpaid, review: () => run(false), approve: () => run(true), cancel, dispose: () => { cancel(); disposed = true } } } diff --git a/neode-ui/src/stores/appLauncher.ts b/neode-ui/src/stores/appLauncher.ts index 958b38e6..8f456acc 100644 --- a/neode-ui/src/stores/appLauncher.ts +++ b/neode-ui/src/stores/appLauncher.ts @@ -608,7 +608,7 @@ export const useAppLauncherStore = defineStore('appLauncher', () => { approveConsent, denyConsent, setNativeIdentityBusy: (busy: boolean) => { nativeIdentityBusy.value = busy }, - rentalRequest: rentalBridge.request, rentalQuote: rentalBridge.quote, rentalPhase: rentalBridge.phase, + rentalRequest: rentalBridge.request, rentalQuote: rentalBridge.quote, rentalPhase: rentalBridge.phase, rentalProgress: rentalBridge.progress, cancelUnpaidRental: rentalBridge.cancelUnpaid, rentalError: rentalBridge.error, reviewRental: rentalBridge.review, approveRental: rentalBridge.approve, cancelRental: rentalBridge.cancel, registrationRequest: registrationBridge.request, registrationPhase: registrationBridge.phase, diff --git a/neode-ui/src/views/AppSession.vue b/neode-ui/src/views/AppSession.vue index b624a029..ddd62134 100644 --- a/neode-ui/src/views/AppSession.vue +++ b/neode-ui/src/views/AppSession.vue @@ -91,7 +91,7 @@ - + diff --git a/neode-ui/src/views/NostrTabSigner.vue b/neode-ui/src/views/NostrTabSigner.vue index 67e62fc6..2532065f 100644 --- a/neode-ui/src/views/NostrTabSigner.vue +++ b/neode-ui/src/views/NostrTabSigner.vue @@ -6,7 +6,7 @@ @select="onIdentitySelected" @cancel="cancelIdentitySelection" /> - +