lfg2025/archy
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases 12 Wiki Activity

Publish content with an atomic price and visibility policy

Browse Source
This commit is contained in:
archipelago
2026-10-06 07:51:00 -04:00
parent 65b51b98f4
commit 19207282f9
9 changed files with 456 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/rpc/content.rs
+152 -92
View File
@@ -19,6 +19,81 @@ fn is_valid_v3_onion(addr: &str) -> bool {
const FILE_CATALOG_PROTOCOL: &str = "https://archipelago.dev/protocols/file-catalog/v1";
fn parse_content_access(params: &serde_json::Value) -> Result<AccessControl> {
let access_type = match params.get("access") {
None => "free",
Some(value) => value.as_str().context("Invalid access type")?,
};
match access_type {
"free" => Ok(AccessControl::Free),
"peers_only" => Ok(AccessControl::PeersOnly),
"paid" => {
let price = params
.get("price_sats")
.and_then(|v| v.as_u64())
.unwrap_or(0);
if price == 0 {
return Err(anyhow::anyhow!("Paid content requires price_sats > 0"));
}
// Optional list of payment methods the sharer accepts.
// Absent/empty = all methods (backward compatible).
const KNOWN_METHODS: [&str; 4] = ["lightning", "onchain", "ecash", "fedimint"];
let accepted = match params.get("accepted_methods") {
None => Vec::new(),
Some(value) => value
.as_array()
.context("Invalid accepted methods")?
.iter()
.map(|method| {
let method = method.as_str().context("Invalid payment method")?;
anyhow::ensure!(
KNOWN_METHODS.contains(&method),
"Unsupported payment method"
);
Ok(method.to_owned())
})
.collect::<Result<Vec<_>>>()?,
};
Ok(AccessControl::Paid {
price_sats: price,
accepted,
})
}
_ => return Err(anyhow::anyhow!("Invalid access type: {}", access_type)),
}
}
fn parse_content_availability(params: &serde_json::Value, default: &str) -> Result<Availability> {
let availability_type = match params.get("availability") {
None => default,
Some(value) => value.as_str().context("Invalid availability")?,
};
match availability_type {
"nobody" => Ok(Availability::Nobody),
"all_peers" => Ok(Availability::AllPeers),
"specific" => {
let peers = params
.get("peers")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect::<Vec<_>>()
})
.unwrap_or_default();
Ok(Availability::Specific { peers })
}
_ => {
return Err(anyhow::anyhow!(
"Invalid availability: {}",
availability_type
))
}
}
}
/// Best-effort reclaim of an ecash payment token that was minted but the sale
/// didn't complete (seller unreachable or couldn't redeem it), so the buyer
/// doesn't lose the value. For Fedimint the spender can reissue its own
@@ -238,6 +313,20 @@ impl RpcHandler {
Ok(serde_json::json!({ "items": catalog.items }))
}
/// Explicit atomic publication endpoint. Older servers reject this method
/// instead of accepting an add request while ignoring its pricing fields.
pub(super) async fn handle_content_publish(
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let policy = params.as_ref().context("Missing params")?;
anyhow::ensure!(
policy.get("access").is_some() && policy.get("availability").is_some(),
"A complete sharing policy is required"
);
self.handle_content_add(params).await
}
/// Add content to my catalog.
pub(super) async fn handle_content_add(
&self,
@@ -282,8 +371,10 @@ impl RpcHandler {
mime_type: mime_type.to_string(),
size_bytes: 0,
description: description.to_string(),
access: AccessControl::Free,
availability: Availability::default(),
access: parse_content_access(&params)?,
// Legacy multi-call clients must configure visibility explicitly;
// an interrupted setup must not publish a paid file as free.
availability: parse_content_availability(&params, "nobody")?,
added_at: chrono::Utc::now().to_rfc3339(),
};
@@ -293,34 +384,45 @@ impl RpcHandler {
item.size_bytes = metadata.len();
}
content_server::add_item(&self.config.data_dir, item.clone()).await?;
let catalog = content_server::add_item(&self.config.data_dir, item.clone()).await?;
let item = catalog
.items
.into_iter()
.find(|saved| saved.filename == item.filename)
.context("Saved content item is unavailable")?;
// Also store as DWN message for interoperable file catalog
if let Ok(store) = DwnStore::new(&self.config.data_dir).await {
let did = crate::identity::did_key_from_pubkey_hex(
&self.state_manager.get_snapshot().await.0.server_info.pubkey,
)
.unwrap_or_default();
let dwn_data = serde_json::json!({
"id": item.id,
"title": item.filename,
"description": item.description,
"content_type": item.mime_type,
"size_bytes": item.size_bytes,
"access": format!("{:?}", item.access).to_lowercase(),
"created_at": item.added_at,
});
if let Err(e) = store
.write_message(
&did,
Some(FILE_CATALOG_PROTOCOL),
Some("https://archipelago.dev/schemas/file-entry/v1"),
Some("application/json"),
Some(dwn_data),
// Export only explicitly public metadata. A staged or peer-restricted
// share must not leak its filename through the public DWN catalog.
if matches!(&item.availability, Availability::AllPeers)
&& !matches!(&item.access, AccessControl::PeersOnly)
{
// Also store as DWN message for interoperable file catalog
if let Ok(store) = DwnStore::new(&self.config.data_dir).await {
let did = crate::identity::did_key_from_pubkey_hex(
&self.state_manager.get_snapshot().await.0.server_info.pubkey,
)
.await
{
debug!("DWN file catalog write (non-fatal): {}", e);
.unwrap_or_default();
let dwn_data = serde_json::json!({
"id": item.id,
"title": item.filename,
"description": item.description,
"content_type": item.mime_type,
"size_bytes": item.size_bytes,
"access": format!("{:?}", item.access).to_lowercase(),
"created_at": item.added_at,
});
if let Err(e) = store
.write_message(
&did,
Some(FILE_CATALOG_PROTOCOL),
Some("https://archipelago.dev/schemas/file-entry/v1"),
Some("application/json"),
Some(dwn_data),
)
.await
{
debug!("DWN file catalog write (non-fatal): {}", e);
}
}
}
@@ -342,6 +444,26 @@ impl RpcHandler {
Ok(serde_json::json!({ "removed": true }))
}
/// Save a complete sharing policy without an intermediate public/free state.
pub(super) async fn handle_content_configure(
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let params = params.context("Missing params")?;
let id = params
.get("id")
.and_then(|v| v.as_str())
.context("Missing id")?;
anyhow::ensure!(
params.get("access").is_some() && params.get("availability").is_some(),
"A complete sharing policy is required"
);
let access = parse_content_access(&params)?;
let availability = parse_content_availability(&params, "nobody")?;
content_server::configure_item(&self.config.data_dir, id, access, availability).await?;
Ok(serde_json::json!({"updated":true}))
}
/// Set pricing for a content item.
pub(super) async fn handle_content_set_pricing(
&self,
@@ -352,43 +474,7 @@ impl RpcHandler {
.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("Missing id"))?;
let access_type = params
.get("access")
.and_then(|v| v.as_str())
.unwrap_or("free");
let access = match access_type {
"free" => AccessControl::Free,
"peers_only" => AccessControl::PeersOnly,
"paid" => {
let price = params
.get("price_sats")
.and_then(|v| v.as_u64())
.unwrap_or(0);
if price == 0 {
return Err(anyhow::anyhow!("Paid content requires price_sats > 0"));
}
// Optional list of payment methods the sharer accepts.
// Absent/empty = all methods (backward compatible).
const KNOWN_METHODS: [&str; 4] = ["lightning", "onchain", "ecash", "fedimint"];
let accepted: Vec<String> = params
.get("accepted_methods")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|m| m.as_str())
.filter(|m| KNOWN_METHODS.contains(m))
.map(str::to_string)
.collect()
})
.unwrap_or_default();
AccessControl::Paid {
price_sats: price,
accepted,
}
}
_ => return Err(anyhow::anyhow!("Invalid access type: {}", access_type)),
};
let access = parse_content_access(&params)?;
content_server::set_access(&self.config.data_dir, id, access).await?;
Ok(serde_json::json!({ "updated": true }))
@@ -404,33 +490,7 @@ impl RpcHandler {
.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("Missing id"))?;
let availability_type = params
.get("availability")
.and_then(|v| v.as_str())
.unwrap_or("all_peers");
let availability = match availability_type {
"nobody" => Availability::Nobody,
"all_peers" => Availability::AllPeers,
"specific" => {
let peers = params
.get("peers")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect::<Vec<_>>()
})
.unwrap_or_default();
Availability::Specific { peers }
}
_ => {
return Err(anyhow::anyhow!(
"Invalid availability: {}",
availability_type
))
}
};
let availability = parse_content_availability(&params, "all_peers")?;
content_server::set_availability(&self.config.data_dir, id, availability).await?;
Ok(serde_json::json!({ "updated": true }))
core/archipelago/src/api/rpc/content_tests.rs
+86
View File
@@ -267,3 +267,89 @@ async fn onchain_delivery_streams_to_owned_cache_and_preserves_incomplete_recove
.is_err()
);
}
#[test]
fn share_creation_stays_hidden_without_explicit_visibility_and_rejects_invalid_policy() {
let empty = serde_json::json!({});
assert!(matches!(
parse_content_availability(&empty, "nobody").unwrap(),
Availability::Nobody
));
assert!(matches!(
parse_content_access(&empty).unwrap(),
AccessControl::Free
));
for invalid in [
serde_json::json!({"access":null}),
serde_json::json!({"access":"paid","price_sats":0}),
serde_json::json!({"access":"paid","price_sats":1,"accepted_methods":["unsupported"]}),
serde_json::json!({"access":"paid","price_sats":1,"accepted_methods":"ecash"}),
] {
assert!(parse_content_access(&invalid).is_err());
}
let paid = serde_json::json!({"access":"paid","price_sats":1,"accepted_methods":["ecash"],"availability":"all_peers"});
assert!(matches!(
parse_content_access(&paid).unwrap(),
AccessControl::Paid { price_sats: 1, .. }
));
assert!(matches!(
parse_content_availability(&paid, "nobody").unwrap(),
Availability::AllPeers
));
}
#[tokio::test]
async fn repeated_share_returns_stable_id_and_complete_policy() {
let dir = tempfile::tempdir().unwrap();
let mut config = crate::config::Config::default();
config.data_dir = dir.path().to_path_buf();
config.dev_mode = false;
let sessions = crate::session::SessionStore::new_for_tests(dir.path().join("sessions.json"));
let handler = RpcHandler::new(
config,
std::sync::Arc::new(crate::state::StateManager::new()),
std::sync::Arc::new(crate::monitoring::MetricsStore::new()),
sessions,
None,
None,
)
.await
.unwrap();
let first = handler
.handle_content_publish(Some(serde_json::json!({
"filename":"film.mp4", "access":"paid", "price_sats":1,
"accepted_methods":["ecash"], "availability":"nobody"
})))
.await
.unwrap();
let second = handler
.handle_content_publish(Some(serde_json::json!({
"filename":"film.mp4", "access":"paid", "price_sats":2,
"accepted_methods":["lightning"], "availability":"nobody"
})))
.await
.unwrap();
assert_eq!(first["item"]["id"], second["item"]["id"]);
let catalog = content_server::load_catalog(dir.path()).await.unwrap();
assert_eq!(catalog.items.len(), 1);
let item = &catalog.items[0];
assert_eq!(second["item"]["id"].as_str(), Some(item.id.as_str()));
assert!(
matches!(&item.access, AccessControl::Paid { price_sats: 2, accepted } if accepted == &["lightning"])
);
assert!(matches!(item.availability, Availability::Nobody));
assert!(handler
.handle_content_configure(Some(serde_json::json!({
"id":item.id, "access":"free"
})))
.await
.is_err());
assert!(matches!(
content_server::load_catalog(dir.path())
.await
.unwrap()
.items[0]
.access,
AccessControl::Paid { price_sats: 2, .. }
));
}
core/archipelago/src/api/rpc/dispatcher.rs
+2
View File
@@ -322,6 +322,8 @@ impl RpcHandler {
// Content catalog management
"content.list-mine" => self.handle_content_list_mine().await,
"content.add" => self.handle_content_add(params).await,
"content.publish" => self.handle_content_publish(params).await,
"content.configure" => self.handle_content_configure(params).await,
"content.remove" => self.handle_content_remove(params).await,
"content.set-pricing" => self.handle_content_set_pricing(params).await,
"content.set-availability" => self.handle_content_set_availability(params).await,
core/archipelago/src/content_server.rs
+135 -17
View File
@@ -6,11 +6,12 @@
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use tokio::fs;
use tokio::{fs, io::AsyncWriteExt, sync::Mutex};
use tracing::{debug, warn};
const CATALOG_FILE: &str = "content/catalog.json";
const CONTENT_DIR: &str = "content/files";
static CATALOG_WRITES: Mutex<()> = Mutex::const_new(());
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ContentItem {
@@ -78,27 +79,39 @@ pub struct ContentCatalog {
/// Load the content catalog from disk.
pub async fn load_catalog(data_dir: &Path) -> Result<ContentCatalog> {
let path = data_dir.join(CATALOG_FILE);
if !path.exists() {
return Ok(ContentCatalog::default());
}
let content = fs::read_to_string(&path)
.await
.context("Failed to read content catalog")?;
let catalog: ContentCatalog = serde_json::from_str(&content).unwrap_or_default();
Ok(catalog)
let content = match fs::read(&path).await {
Ok(bytes) => bytes,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Ok(ContentCatalog::default())
}
Err(error) => return Err(error).context("Failed to read content catalog"),
};
serde_json::from_slice(&content)
.context("Invalid content catalog; existing shares were preserved")
}
/// Save the content catalog to disk.
pub async fn save_catalog(data_dir: &Path, catalog: &ContentCatalog) -> Result<()> {
let _lock = CATALOG_WRITES.lock().await;
save_catalog_unlocked(data_dir, catalog).await
}
async fn save_catalog_unlocked(data_dir: &Path, catalog: &ContentCatalog) -> Result<()> {
let dir = data_dir.join("content");
fs::create_dir_all(&dir)
.await
.context("Failed to create content dir")?;
let path = data_dir.join(CATALOG_FILE);
let content = serde_json::to_string_pretty(catalog).context("Failed to serialize catalog")?;
fs::write(&path, content)
.await
.context("Failed to write catalog")?;
let (file, temporary) = tempfile::NamedTempFile::new_in(&dir)?.into_parts();
let mut file = fs::File::from_std(file);
file.write_all(content.as_bytes()).await?;
file.sync_all().await?;
drop(file);
// Complete the rename synchronously while holding the catalog lock: an
// aborted async caller must not leave a late rename racing the next writer.
std::fs::rename(&temporary, &path).context("Failed to commit content catalog")?;
fs::File::open(&dir).await?.sync_all().await?;
Ok(())
}
@@ -106,13 +119,14 @@ pub async fn save_catalog(data_dir: &Path, catalog: &ContentCatalog) -> Result<(
/// means the entry gets pruned again next time it's requested, so errors are
/// logged rather than propagated.
async fn prune_missing_content_entry(data_dir: &Path, id: &str) {
let _lock = CATALOG_WRITES.lock().await;
let Ok(mut catalog) = load_catalog(data_dir).await else {
return;
};
let before = catalog.items.len();
catalog.items.retain(|i| i.id != id);
if catalog.items.len() != before {
if let Err(e) = save_catalog(data_dir, &catalog).await {
if let Err(e) = save_catalog_unlocked(data_dir, &catalog).await {
warn!(error = %e, content_id = %id, "failed to save catalog after pruning missing content entry");
}
}
@@ -149,6 +163,7 @@ pub fn content_file_path(data_dir: &Path, item: &ContentItem) -> PathBuf {
/// (2026-07-22). Same filename → update the existing entry in place and
/// keep its id, so existing buyers' owned records stay valid.
pub async fn add_item(data_dir: &Path, item: ContentItem) -> Result<ContentCatalog> {
let _lock = CATALOG_WRITES.lock().await;
let mut catalog = load_catalog(data_dir).await?;
if catalog.items.iter().any(|i| i.id == item.id) {
return Err(anyhow::anyhow!("Content item '{}' already exists", item.id));
@@ -165,24 +180,26 @@ pub async fn add_item(data_dir: &Path, item: ContentItem) -> Result<ContentCatal
} else {
catalog.items.push(item);
}
save_catalog(data_dir, &catalog).await?;
save_catalog_unlocked(data_dir, &catalog).await?;
Ok(catalog)
}
/// Remove a content item from the catalog.
pub async fn remove_item(data_dir: &Path, id: &str) -> Result<ContentCatalog> {
let _lock = CATALOG_WRITES.lock().await;
let mut catalog = load_catalog(data_dir).await?;
catalog.items.retain(|i| i.id != id);
save_catalog(data_dir, &catalog).await?;
save_catalog_unlocked(data_dir, &catalog).await?;
Ok(catalog)
}
/// Update access control for a content item.
pub async fn set_access(data_dir: &Path, id: &str, access: AccessControl) -> Result<()> {
let _lock = CATALOG_WRITES.lock().await;
let mut catalog = load_catalog(data_dir).await?;
if let Some(item) = catalog.items.iter_mut().find(|i| i.id == id) {
item.access = access;
save_catalog(data_dir, &catalog).await?;
save_catalog_unlocked(data_dir, &catalog).await?;
Ok(())
} else {
Err(anyhow::anyhow!("Content item '{}' not found", id))
@@ -191,16 +208,36 @@ pub async fn set_access(data_dir: &Path, id: &str, access: AccessControl) -> Res
/// Update availability for a content item.
pub async fn set_availability(data_dir: &Path, id: &str, availability: Availability) -> Result<()> {
let _lock = CATALOG_WRITES.lock().await;
let mut catalog = load_catalog(data_dir).await?;
if let Some(item) = catalog.items.iter_mut().find(|i| i.id == id) {
item.availability = availability;
save_catalog(data_dir, &catalog).await?;
save_catalog_unlocked(data_dir, &catalog).await?;
Ok(())
} else {
Err(anyhow::anyhow!("Content item '{}' not found", id))
}
}
/// Change price and visibility in one durable catalog transaction.
pub async fn configure_item(
data_dir: &Path,
id: &str,
access: AccessControl,
availability: Availability,
) -> Result<()> {
let _lock = CATALOG_WRITES.lock().await;
let mut catalog = load_catalog(data_dir).await?;
let item = catalog
.items
.iter_mut()
.find(|item| item.id == id)
.context("Content item not found")?;
item.access = access;
item.availability = availability;
save_catalog_unlocked(data_dir, &catalog).await
}
/// A byte range request (start, optional end).
pub enum ByteRange {
From { start: u64, end: Option<u64> },
@@ -1066,6 +1103,87 @@ mod paid_read_order_tests {
dir
}
#[tokio::test]
async fn corrupt_catalog_is_not_overwritten_by_a_mutation() {
let dir = fixture(b"file").await;
let item = load_catalog(dir.path()).await.unwrap().items.remove(0);
fs::write(dir.path().join(CATALOG_FILE), b"corrupt")
.await
.unwrap();
assert!(add_item(dir.path(), item).await.is_err());
assert!(remove_item(dir.path(), "paid").await.is_err());
assert_eq!(
fs::read(dir.path().join(CATALOG_FILE)).await.unwrap(),
b"corrupt"
);
}
#[tokio::test]
async fn concurrent_catalog_updates_preserve_all_items_and_sharing_fields() {
let dir = fixture(b"file").await;
let template = load_catalog(dir.path()).await.unwrap().items.remove(0);
let mut tasks = Vec::new();
for i in 0..16 {
let root = dir.path().to_owned();
let mut item = template.clone();
item.id = format!("item-{i}");
item.filename = format!("file-{i}");
tasks.push(tokio::spawn(async move {
add_item(&root, item).await.unwrap();
}));
}
for task in tasks {
task.await.unwrap();
}
assert_eq!(load_catalog(dir.path()).await.unwrap().items.len(), 17);
let (price, visibility) = tokio::join!(
set_access(
dir.path(),
"paid",
AccessControl::Paid {
price_sats: 7,
accepted: vec!["ecash".into()]
}
),
set_availability(dir.path(), "paid", Availability::Nobody),
);
price.unwrap();
visibility.unwrap();
let item = load_catalog(dir.path()).await.unwrap().items.remove(0);
assert!(matches!(
item.access,
AccessControl::Paid { price_sats: 7, .. }
));
assert!(matches!(item.availability, Availability::Nobody));
configure_item(
dir.path(),
"paid",
AccessControl::Paid {
price_sats: 9,
accepted: vec!["ecash".into()],
},
Availability::AllPeers,
)
.await
.unwrap();
let item = load_catalog(dir.path()).await.unwrap().items.remove(0);
assert!(matches!(
item.access,
AccessControl::Paid { price_sats: 9, .. }
));
assert!(matches!(item.availability, Availability::AllPeers));
use std::os::unix::fs::PermissionsExt;
assert_eq!(
fs::metadata(dir.path().join(CATALOG_FILE))
.await
.unwrap()
.permissions()
.mode()
& 0o777,
0o600
);
}
#[tokio::test]
async fn all_read_failures_precede_redemption_even_as_root() {
for kind in [