cdricms/vnidrop
1
1
Fork 0
mirror of https://github.com/sudosylabs/vnidrop.git synced 2026-08-05 02:29:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Rust core architecture

Browse Source
This commit is contained in:
Hammed Abass
2026-07-03 00:09:11 +02:00
parent 90463d7210
commit 3da576a94e
21 changed files with 2072 additions and 590 deletions
Download Patch File Download Diff File Expand all files Collapse all files

811
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

2
Cargo.toml
View File

@@ -1,5 +1,5 @@
[workspace]
members = ["crates/vnidrop-core"]
members = ["crates/vnidrop"]
resolver = "2"
[profile.release]

8
crates/vnidrop-core/Cargo.toml → crates/vnidrop/Cargo.toml
View File

@@ -1,10 +1,10 @@
[package]
name = "vnidrop-core"
name = "vnidrop"
version = "0.1.0"
edition = "2021"
[lib]
name = "vnidrop_core"
name = "vnidrop"
crate-type = ["cdylib", "staticlib", "rlib"]
[dependencies]
@@ -21,7 +21,11 @@ n0-future = "0.3.1"
num_cpus = "1.17.0"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8.6", default-features = false, features = ["runtime-tokio", "sqlite", "macros", "migrate"] }
thiserror = "2.0.18"
tracing = "0.1.41"
tracing-appender = "0.2.4"
tracing-subscriber = { version = "0.3.20", features = ["env-filter", "fmt"] }
tokio = { version = "1.52.3", features = ["full"] }
uniffi = { version = "=0.29.4", features = ["tokio"] }
uuid = { version = "1.23.3", features = ["v4", "serde"] }

81
crates/vnidrop/src/access_policy.rs Normal file
View File

@@ -0,0 +1,81 @@
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use tokio::sync::RwLock;
use crate::api::TransferAccessMode;
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum AccessDecision {
Allow,
Deny { reason: &'static str },
}
#[derive(Debug, Default)]
pub(crate) struct AccessPolicy {
modes: RwLock<HashMap<u64, TransferAccessMode>>,
approved_sessions: RwLock<HashSet<(u64, String)>>,
}
impl AccessPolicy {
pub(crate) fn new() -> Arc<Self> {
Arc::new(Self::default())
}
pub(crate) async fn set_mode(&self, transfer_id: u64, mode: TransferAccessMode) {
self.modes.write().await.insert(transfer_id, mode);
}
pub(crate) async fn remove_transfer(&self, transfer_id: u64) {
self.modes.write().await.remove(&transfer_id);
self.approved_sessions
.write()
.await
.retain(|(id, _)| *id != transfer_id);
}
pub(crate) async fn approve_endpoint(&self, transfer_id: u64, endpoint_id: String) {
self.approved_sessions
.write()
.await
.insert((transfer_id, endpoint_id));
}
pub(crate) async fn decide(
&self,
transfer_id: u64,
endpoint_id: Option<&str>,
) -> AccessDecision {
match self
.modes
.read()
.await
.get(&transfer_id)
.cloned()
.unwrap_or(TransferAccessMode::Public)
{
TransferAccessMode::Public => AccessDecision::Allow,
TransferAccessMode::ApprovalRequired => {
let Some(endpoint_id) = endpoint_id else {
return AccessDecision::Deny {
reason: "missing-endpoint-id",
};
};
if self
.approved_sessions
.read()
.await
.contains(&(transfer_id, endpoint_id.to_string()))
{
AccessDecision::Allow
} else {
AccessDecision::Deny {
reason: "approval-required",
}
}
}
}
}
}

123
crates/vnidrop/src/api.rs Normal file
View File

@@ -0,0 +1,123 @@
use iroh_blobs::Hash;
use serde::{Deserialize, Serialize};
use crate::util::{non_empty, now_ms};
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct CoreEvent {
pub id: String,
pub timestamp: i64,
pub scope: String,
pub transfer_id: Option<u64>,
pub direction: Option<String>,
pub phase: String,
pub kind: String,
pub data_json: String,
}
#[uniffi::export(with_foreign)]
pub trait CoreEventSink: Send + Sync {
fn on_event(&self, event: CoreEvent);
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct RuntimeStatus {
pub endpoint_id: String,
pub addr: String,
pub active_transfers: u64,
pub active_shares: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Enum)]
pub enum SourceKind {
Path,
AndroidContentUri,
IosSecurityScopedUrl,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct ShareSource {
pub kind: SourceKind,
pub value: String,
pub display_name: Option<String>,
pub is_directory: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct ShareMetadataInput {
pub transfer_id: u64,
pub transfer_name: Option<String>,
pub sender_name: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Enum)]
pub enum TransferAccessMode {
Public,
ApprovalRequired,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct StoredTransfer {
pub transfer_id: u64,
pub direction: String,
pub status: String,
pub transfer_name: Option<String>,
pub content_hash: Option<String>,
pub ticket: Option<String>,
pub file_count: u64,
pub total_size: u64,
pub created_at: i64,
pub updated_at: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct ShareResult {
pub transfer_id: u64,
pub ticket: String,
pub blob_ticket: String,
pub hash: String,
pub transfer_name: String,
pub file_count: u64,
pub total_size: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct TransferMetadata {
pub version: u8,
pub transfer_id: u64,
pub transfer_name: String,
pub sender_name: Option<String>,
pub created_at: i64,
pub content_hash: String,
pub file_count: u64,
pub total_size: u64,
}
impl TransferMetadata {
pub(crate) fn new(
transfer_id: u64,
transfer_name: impl Into<String>,
sender_name: Option<String>,
content_hash: Hash,
file_count: u64,
total_size: u64,
) -> Self {
Self {
version: 1,
transfer_id,
transfer_name: transfer_name.into(),
sender_name: sender_name.and_then(non_empty),
created_at: now_ms(),
content_hash: content_hash.to_string(),
file_count,
total_size,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct TicketInspection {
pub kind: String,
pub blob_ticket: String,
pub metadata: Option<TransferMetadata>,
}

23
crates/vnidrop/src/error.rs Normal file
View File

@@ -0,0 +1,23 @@
use std::io;
#[derive(Debug, thiserror::Error, uniffi::Error)]
pub enum VnidropError {
#[error("{reason}")]
Generic { reason: String },
}
impl From<anyhow::Error> for VnidropError {
fn from(error: anyhow::Error) -> Self {
Self::Generic {
reason: error.to_string(),
}
}
}
impl From<io::Error> for VnidropError {
fn from(error: io::Error) -> Self {
Self::Generic {
reason: error.to_string(),
}
}
}

227
crates/vnidrop/src/filesystem.rs Normal file
View File

@@ -0,0 +1,227 @@
use std::{
io::{self, Read, Write},
path::{Component, Path, PathBuf},
};
use anyhow::{Context, Result};
use bytes::Bytes;
use iroh_blobs::{api::TempTag, Hash};
use crate::{
api::{ShareSource, SourceKind},
util::non_empty,
};
const STREAM_BUFFER_LEN: usize = 1024 * 1024;
#[derive(Debug)]
pub(crate) struct TransferImport {
pub(crate) tag: TempTag,
pub(crate) root_hash: Hash,
pub(crate) total_size: u64,
pub(crate) file_count: u64,
pub(crate) default_name: String,
}
#[derive(Debug)]
pub(crate) struct ImportSourceFile {
pub(crate) path: PathBuf,
pub(crate) collection_name: String,
}
pub(crate) fn collect_import_files(sources: Vec<ShareSource>) -> Result<Vec<ImportSourceFile>> {
let mut files = Vec::new();
for source in sources {
match source.kind {
SourceKind::Path | SourceKind::IosSecurityScopedUrl => {
let path = source_path(&source)?;
let display_name = source
.display_name
.clone()
.and_then(non_empty)
.or_else(|| {
path.file_name()
.and_then(|name| name.to_str())
.map(ToOwned::to_owned)
})
.unwrap_or_else(|| "transfer".to_string());
if source.is_directory || path.is_dir() {
collect_dir_files(&path, &display_name, &mut files)?;
} else {
files.push(ImportSourceFile {
path,
collection_name: validated_relative_string(&display_name)?,
});
}
}
SourceKind::AndroidContentUri => {
anyhow::bail!(
"Android content URI streaming needs platform file descriptor glue before it can be imported without copying"
);
}
}
}
if files.is_empty() {
anyhow::bail!("no files found in selected sources");
}
Ok(files)
}
fn source_path(source: &ShareSource) -> Result<PathBuf> {
if matches!(source.kind, SourceKind::IosSecurityScopedUrl)
&& source.value.starts_with("file://")
{
let without_scheme = source.value.trim_start_matches("file://");
return Ok(PathBuf::from(percent_decode_file_url_path(without_scheme)?));
}
Ok(PathBuf::from(&source.value))
}
fn collect_dir_files(
root: &Path,
display_name: &str,
files: &mut Vec<ImportSourceFile>,
) -> Result<()> {
for entry in walkdir::WalkDir::new(root).follow_links(false) {
let entry = entry?;
if !entry.file_type().is_file() {
continue;
}
let relative = entry
.path()
.strip_prefix(root)
.context("failed to compute relative path")?;
let collection_name = path_to_string(Path::new(display_name).join(relative), true)?;
files.push(ImportSourceFile {
path: entry.path().to_path_buf(),
collection_name,
});
}
Ok(())
}
pub(crate) fn default_collection_name(files: &[ImportSourceFile]) -> String {
files
.first()
.and_then(|file| file.collection_name.split('/').next())
.filter(|name| !name.is_empty())
.unwrap_or("transfer")
.to_string()
}
pub(crate) fn safe_output_path(output_dir: &Path, relative_path: &str) -> Result<PathBuf> {
let relative = Path::new(relative_path);
path_to_string(relative, true)?;
Ok(output_dir.join(relative))
}
pub(crate) fn read_stream_from_blocking_reader<R>(
mut reader: R,
) -> impl futures::Stream<Item = io::Result<Bytes>> + Send + Sync + 'static
where
R: Read + Send + 'static,
{
let (tx, rx) = async_channel::bounded(2);
std::thread::spawn(move || {
let mut buffer = vec![0; STREAM_BUFFER_LEN];
loop {
match reader.read(&mut buffer) {
Ok(0) => break,
Ok(read) => {
if tx
.send_blocking(Ok(Bytes::copy_from_slice(&buffer[..read])))
.is_err()
{
break;
}
}
Err(error) => {
let _ = tx.send_blocking(Err(error));
break;
}
}
}
});
rx
}
pub(crate) fn write_stream_to_blocking_writer<W>(
mut writer: W,
rx: async_channel::Receiver<io::Result<Option<Bytes>>>,
) -> io::Result<()>
where
W: Write,
{
while let Ok(item) = rx.recv_blocking() {
match item? {
Some(bytes) => writer.write_all(&bytes)?,
None => break,
}
}
writer.flush()
}
pub(crate) async fn wait_for_writer(
task: std::thread::JoinHandle<io::Result<()>>,
) -> Result<io::Result<()>> {
tokio::task::spawn_blocking(move || {
task.join()
.map_err(|_| anyhow::anyhow!("export writer thread panicked"))
})
.await?
}
pub(crate) fn validated_relative_string(name: &str) -> Result<String> {
path_to_string(Path::new(name), true)
}
pub(crate) fn path_to_string(path: impl AsRef<Path>, must_be_relative: bool) -> Result<String> {
let mut path_str = String::new();
let parts = path
.as_ref()
.components()
.filter_map(|component| match component {
Component::Normal(x) => {
let Some(component) = x.to_str() else {
return Some(Err(anyhow::anyhow!("invalid character in path")));
};
if !component.contains('/') && !component.contains('\\') {
Some(Ok(component))
} else {
Some(Err(anyhow::anyhow!("invalid path component {component:?}")))
}
}
Component::RootDir => {
if must_be_relative {
Some(Err(anyhow::anyhow!("invalid root path component")))
} else {
path_str.push('/');
None
}
}
other => Some(Err(anyhow::anyhow!("invalid path component {other:?}"))),
})
.collect::<Result<Vec<_>>>()?;
path_str.push_str(&parts.join("/"));
Ok(path_str)
}
pub(crate) fn percent_decode_file_url_path(value: &str) -> Result<String> {
let bytes = value.as_bytes();
let mut output = Vec::with_capacity(bytes.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'%' {
if i + 2 >= bytes.len() {
anyhow::bail!("invalid percent escape in file URL");
}
let hex = std::str::from_utf8(&bytes[i + 1..i + 3])?;
output.push(u8::from_str_radix(hex, 16).context("invalid percent escape in file URL")?);
i += 3;
} else {
output.push(bytes[i]);
i += 1;
}
}
Ok(String::from_utf8(output)?)
}

22
crates/vnidrop/src/lib.rs Normal file
View File

@@ -0,0 +1,22 @@
mod access_policy;
mod api;
mod error;
mod filesystem;
mod logging;
mod repository;
mod runtime;
mod secret;
mod ticket;
mod util;
pub use api::{
CoreEvent, CoreEventSink, RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource,
SourceKind, StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata,
};
pub use error::VnidropError;
pub use runtime::VnidropCore;
uniffi::setup_scaffolding!();
#[cfg(test)]
mod tests;

29
crates/vnidrop/src/logging.rs Normal file
View File

@@ -0,0 +1,29 @@
use std::{path::Path, sync::OnceLock};
use anyhow::Result;
use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter};
static LOG_GUARD: OnceLock<tracing_appender::non_blocking::WorkerGuard> = OnceLock::new();
pub(crate) fn init_logging(app_data_dir: &Path) -> Result<()> {
if LOG_GUARD.get().is_some() {
return Ok(());
}
let log_dir = app_data_dir.join("logs");
std::fs::create_dir_all(&log_dir)?;
let file_appender = tracing_appender::rolling::daily(log_dir, "vnidrop.log");
let (writer, guard) = tracing_appender::non_blocking(file_appender);
let filter = EnvFilter::try_from_default_env()
.unwrap_or_else(|_| EnvFilter::new("vnidrop=debug,iroh=info,iroh_blobs=info,warn"));
let subscriber = tracing_subscriber::registry()
.with(filter)
.with(fmt::layer().with_writer(writer).with_ansi(false));
if tracing::subscriber::set_global_default(subscriber).is_ok() {
let _ = LOG_GUARD.set(guard);
}
Ok(())
}

228
crates/vnidrop/src/repository.rs Normal file
View File

@@ -0,0 +1,228 @@
use std::{path::Path, str::FromStr};
use anyhow::Result;
use sqlx::{
sqlite::{SqliteConnectOptions, SqlitePoolOptions},
Row, SqlitePool,
};
use crate::api::{CoreEvent, StoredTransfer};
use crate::util::now_ms;
#[derive(Debug, Clone)]
pub(crate) struct Repository {
pool: SqlitePool,
}
impl Repository {
pub(crate) async fn open(app_data_dir: &Path) -> Result<Self> {
let db_path = app_data_dir.join("vnidrop.sqlite3");
let options = SqliteConnectOptions::from_str("sqlite://")?
.filename(db_path)
.create_if_missing(true);
let pool = SqlitePoolOptions::new()
.max_connections(4)
.connect_with(options)
.await?;
let repository = Self { pool };
repository.ensure_schema().await?;
Ok(repository)
}
async fn ensure_schema(&self) -> Result<()> {
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS transfers (
transfer_id INTEGER PRIMARY KEY,
direction TEXT NOT NULL,
status TEXT NOT NULL,
transfer_name TEXT,
content_hash TEXT,
ticket TEXT,
file_count INTEGER NOT NULL DEFAULT 0,
total_size INTEGER NOT NULL DEFAULT 0,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
"#,
)
.execute(&self.pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS transfer_events (
id TEXT PRIMARY KEY,
timestamp INTEGER NOT NULL,
scope TEXT NOT NULL,
transfer_id INTEGER,
direction TEXT,
phase TEXT NOT NULL,
kind TEXT NOT NULL,
data_json TEXT NOT NULL
);
"#,
)
.execute(&self.pool)
.await?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_transfer_events_transfer_id ON transfer_events(transfer_id, timestamp);",
)
.execute(&self.pool)
.await?;
Ok(())
}
pub(crate) async fn upsert_transfer(
&self,
transfer_id: u64,
direction: &str,
status: &str,
transfer_name: Option<&str>,
content_hash: Option<&str>,
ticket: Option<&str>,
file_count: u64,
total_size: u64,
) -> Result<()> {
let now = now_ms();
sqlx::query(
r#"
INSERT INTO transfers (
transfer_id, direction, status, transfer_name, content_hash, ticket,
file_count, total_size, created_at, updated_at
)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)
ON CONFLICT(transfer_id) DO UPDATE SET
direction = excluded.direction,
status = excluded.status,
transfer_name = excluded.transfer_name,
content_hash = excluded.content_hash,
ticket = excluded.ticket,
file_count = excluded.file_count,
total_size = excluded.total_size,
updated_at = excluded.updated_at;
"#,
)
.bind(transfer_id as i64)
.bind(direction)
.bind(status)
.bind(transfer_name)
.bind(content_hash)
.bind(ticket)
.bind(file_count as i64)
.bind(total_size as i64)
.bind(now)
.execute(&self.pool)
.await?;
Ok(())
}
pub(crate) async fn update_transfer_status(
&self,
transfer_id: u64,
status: &str,
) -> Result<()> {
sqlx::query("UPDATE transfers SET status = ?1, updated_at = ?2 WHERE transfer_id = ?3")
.bind(status)
.bind(now_ms())
.bind(transfer_id as i64)
.execute(&self.pool)
.await?;
Ok(())
}
pub(crate) async fn insert_event(&self, event: &CoreEvent) -> Result<()> {
sqlx::query(
r#"
INSERT OR REPLACE INTO transfer_events (
id, timestamp, scope, transfer_id, direction, phase, kind, data_json
)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8);
"#,
)
.bind(&event.id)
.bind(event.timestamp)
.bind(&event.scope)
.bind(event.transfer_id.map(|value| value as i64))
.bind(&event.direction)
.bind(&event.phase)
.bind(&event.kind)
.bind(&event.data_json)
.execute(&self.pool)
.await?;
Ok(())
}
pub(crate) async fn list_transfers(&self) -> Result<Vec<StoredTransfer>> {
let rows = sqlx::query(
r#"
SELECT transfer_id, direction, status, transfer_name, content_hash, ticket,
file_count, total_size, created_at, updated_at
FROM transfers
ORDER BY updated_at DESC
"#,
)
.fetch_all(&self.pool)
.await?;
Ok(rows.into_iter().map(row_to_transfer).collect())
}
pub(crate) async fn list_events(&self, transfer_id: Option<u64>) -> Result<Vec<CoreEvent>> {
let rows = if let Some(transfer_id) = transfer_id {
sqlx::query(
r#"
SELECT id, timestamp, scope, transfer_id, direction, phase, kind, data_json
FROM transfer_events
WHERE transfer_id = ?1
ORDER BY timestamp ASC
"#,
)
.bind(transfer_id as i64)
.fetch_all(&self.pool)
.await?
} else {
sqlx::query(
r#"
SELECT id, timestamp, scope, transfer_id, direction, phase, kind, data_json
FROM transfer_events
ORDER BY timestamp DESC
LIMIT 500
"#,
)
.fetch_all(&self.pool)
.await?
};
Ok(rows.into_iter().map(row_to_event).collect())
}
}
fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> StoredTransfer {
StoredTransfer {
transfer_id: row.get::<i64, _>("transfer_id") as u64,
direction: row.get("direction"),
status: row.get("status"),
transfer_name: row.get("transfer_name"),
content_hash: row.get("content_hash"),
ticket: row.get("ticket"),
file_count: row.get::<i64, _>("file_count") as u64,
total_size: row.get::<i64, _>("total_size") as u64,
created_at: row.get("created_at"),
updated_at: row.get("updated_at"),
}
}
fn row_to_event(row: sqlx::sqlite::SqliteRow) -> CoreEvent {
CoreEvent {
id: row.get("id"),
timestamp: row.get("timestamp"),
scope: row.get("scope"),
transfer_id: row
.get::<Option<i64>, _>("transfer_id")
.map(|value| value as u64),
direction: row.get("direction"),
phase: row.get("phase"),
kind: row.get("kind"),
data_json: row.get("data_json"),
}
}

797
crates/vnidrop-core/src/lib.rs → crates/vnidrop/src/runtime.rs
View File

@@ -1,29 +1,17 @@
use std::{
collections::HashMap,
fs::File,
io::{self, Read, Write},
path::{Component, Path, PathBuf},
str::FromStr,
io,
path::{Path, PathBuf},
sync::{Arc, Mutex},
time::{SystemTime, UNIX_EPOCH},
};
use anyhow::{Context, Result};
use bytes::Bytes;
use data_encoding::{BASE64URL_NOPAD, HEXLOWER};
use futures_lite::StreamExt as _;
use iroh::{
endpoint::presets,
protocol::Router,
Endpoint, SecretKey,
};
use iroh::{endpoint::presets, protocol::Router, Endpoint};
use iroh_blobs::{
api::{
blobs::AddProgressItem,
proto::ExportRangesItem,
remote::GetProgressItem,
TempTag,
},
api::{blobs::AddProgressItem, proto::ExportRangesItem, remote::GetProgressItem, TempTag},
format::collection::Collection,
get::request::get_hash_seq_and_sizes,
provider::events::{EventMask, EventSender, ProviderMessage, RequestUpdate},
@@ -32,207 +20,26 @@ use iroh_blobs::{
BlobFormat, BlobsProtocol, Hash,
};
use n0_future::BufferedStreamExt;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::sync::{mpsc, oneshot, Mutex as TokioMutex};
const VNIDROP_TICKET_PREFIX: &str = "vnd1:";
const STREAM_BUFFER_LEN: usize = 1024 * 1024;
#[derive(Debug, thiserror::Error, uniffi::Error)]
pub enum VnidropError {
#[error("{reason}")]
Generic { reason: String },
}
impl From<anyhow::Error> for VnidropError {
fn from(error: anyhow::Error) -> Self {
Self::Generic {
reason: error.to_string(),
}
}
}
impl From<io::Error> for VnidropError {
fn from(error: io::Error) -> Self {
Self::Generic {
reason: error.to_string(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct CoreEvent {
pub id: String,
pub timestamp: i64,
pub scope: String,
pub transfer_id: Option<u64>,
pub direction: Option<String>,
pub phase: String,
pub kind: String,
pub data_json: String,
}
#[uniffi::export(with_foreign)]
pub trait CoreEventSink: Send + Sync {
fn on_event(&self, event: CoreEvent);
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct RuntimeStatus {
pub endpoint_id: String,
pub addr: String,
pub active_transfers: u64,
pub active_shares: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Enum)]
pub enum SourceKind {
Path,
AndroidContentUri,
IosSecurityScopedUrl,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct ShareSource {
pub kind: SourceKind,
pub value: String,
pub display_name: Option<String>,
pub is_directory: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct ShareMetadataInput {
pub transfer_id: u64,
pub transfer_name: Option<String>,
pub sender_name: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct ShareResult {
pub transfer_id: u64,
pub ticket: String,
pub blob_ticket: String,
pub hash: String,
pub transfer_name: String,
pub file_count: u64,
pub total_size: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct TransferMetadata {
pub version: u8,
pub transfer_id: u64,
pub transfer_name: String,
pub sender_name: Option<String>,
pub created_at: i64,
pub content_hash: String,
pub file_count: u64,
pub total_size: u64,
}
impl TransferMetadata {
fn new(
transfer_id: u64,
transfer_name: impl Into<String>,
sender_name: Option<String>,
content_hash: Hash,
file_count: u64,
total_size: u64,
) -> Self {
Self {
version: 1,
transfer_id,
transfer_name: transfer_name.into(),
sender_name: sender_name.and_then(non_empty),
created_at: now_ms(),
content_hash: content_hash.to_string(),
file_count,
total_size,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct VnidropTicket {
version: u8,
blob_ticket: String,
metadata: TransferMetadata,
}
impl VnidropTicket {
fn new(blob_ticket: BlobTicket, metadata: TransferMetadata) -> Self {
Self {
version: 1,
blob_ticket: blob_ticket.to_string(),
metadata,
}
}
fn encode(&self) -> Result<String> {
let bytes = serde_json::to_vec(self)?;
Ok(format!(
"{VNIDROP_TICKET_PREFIX}{}",
BASE64URL_NOPAD.encode(&bytes)
))
}
fn decode(value: &str) -> Result<Self> {
let encoded = value
.strip_prefix(VNIDROP_TICKET_PREFIX)
.context("not a VniDrop ticket")?;
let bytes = BASE64URL_NOPAD
.decode(encoded.as_bytes())
.context("invalid VniDrop ticket encoding")?;
serde_json::from_slice(&bytes).context("invalid VniDrop ticket payload")
}
}
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct TicketInspection {
pub kind: String,
pub blob_ticket: String,
pub metadata: Option<TransferMetadata>,
}
#[derive(Debug, Clone)]
struct ParsedTransferTicket {
blob_ticket: BlobTicket,
metadata: Option<TransferMetadata>,
}
fn parse_transfer_ticket(value: &str) -> Result<ParsedTransferTicket> {
if value.starts_with(VNIDROP_TICKET_PREFIX) {
let ticket = VnidropTicket::decode(value)?;
let blob_ticket = BlobTicket::from_str(&ticket.blob_ticket)
.context("invalid BlobTicket inside VniDrop ticket")?;
return Ok(ParsedTransferTicket {
blob_ticket,
metadata: Some(ticket.metadata),
});
}
let blob_ticket = BlobTicket::from_str(value).context("invalid BlobTicket")?;
Ok(ParsedTransferTicket {
blob_ticket,
metadata: None,
})
}
#[derive(Debug)]
struct TransferImport {
tag: TempTag,
root_hash: Hash,
total_size: u64,
file_count: u64,
default_name: String,
}
#[derive(Debug)]
struct ImportSourceFile {
path: PathBuf,
collection_name: String,
}
use crate::{
access_policy::{AccessDecision, AccessPolicy},
api::{
CoreEvent, CoreEventSink, RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource,
StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata,
},
error::VnidropError,
filesystem::{
collect_import_files, default_collection_name, read_stream_from_blocking_reader,
safe_output_path, wait_for_writer, write_stream_to_blocking_writer, TransferImport,
},
logging::init_logging,
repository::Repository,
secret::load_or_create_secret,
ticket::{parse_transfer_ticket, ParsedTransferTicket, VnidropTicket},
util::{non_empty, now_ms, unique_transfer_id},
};
#[derive(uniffi::Object)]
pub struct VnidropCore {
@@ -244,10 +51,13 @@ struct CoreInner {
endpoint: Endpoint,
router: Router,
store: FsStore,
repository: Repository,
access_policy: Arc<AccessPolicy>,
event_sink: Arc<dyn CoreEventSink>,
active_transfers: TokioMutex<HashMap<u64, oneshot::Sender<()>>>,
active_shares: TokioMutex<HashMap<u64, TempTag>>,
hash_to_transfer: TokioMutex<HashMap<String, u64>>,
connection_endpoints: TokioMutex<HashMap<u64, String>>,
sequence: Mutex<u64>,
}
@@ -260,7 +70,7 @@ impl VnidropCore {
) -> Result<Arc<Self>, VnidropError> {
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.thread_name("vnidrop-core")
.thread_name("vnidrop")
.build()?;
let app_data_dir = PathBuf::from(app_data_dir);
let inner = runtime.block_on(CoreInner::start(app_data_dir, event_sink))?;
@@ -288,7 +98,10 @@ impl VnidropCore {
receiver_name: Option<String>,
) -> Result<(), VnidropError> {
self.runtime
.block_on(self.inner.receive(ticket, PathBuf::from(output_dir), receiver_name))
.block_on(
self.inner
.receive(ticket, PathBuf::from(output_dir), receiver_name),
)
.map_err(Into::into)
}
@@ -298,6 +111,41 @@ impl VnidropCore {
.map_err(Into::into)
}
pub fn set_transfer_access_mode(
&self,
transfer_id: u64,
mode: TransferAccessMode,
) -> Result<(), VnidropError> {
self.runtime
.block_on(self.inner.set_transfer_access_mode(transfer_id, mode))
.map_err(Into::into)
}
pub fn approve_endpoint_for_transfer(
&self,
transfer_id: u64,
endpoint_id: String,
) -> Result<(), VnidropError> {
self.runtime
.block_on(
self.inner
.approve_endpoint_for_transfer(transfer_id, endpoint_id),
)
.map_err(Into::into)
}
pub fn list_transfers(&self) -> Result<Vec<StoredTransfer>, VnidropError> {
self.runtime
.block_on(self.inner.repository.list_transfers())
.map_err(Into::into)
}
pub fn list_events(&self, transfer_id: Option<u64>) -> Result<Vec<CoreEvent>, VnidropError> {
self.runtime
.block_on(self.inner.repository.list_events(transfer_id))
.map_err(Into::into)
}
pub fn inspect_ticket(&self, ticket: String) -> Result<TicketInspection, VnidropError> {
let parsed = parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")?;
Ok(TicketInspection {
@@ -317,12 +165,11 @@ impl VnidropCore {
}
impl CoreInner {
async fn start(
app_data_dir: PathBuf,
event_sink: Arc<dyn CoreEventSink>,
) -> Result<Arc<Self>> {
async fn start(app_data_dir: PathBuf, event_sink: Arc<dyn CoreEventSink>) -> Result<Arc<Self>> {
tokio::fs::create_dir_all(&app_data_dir).await?;
init_logging(&app_data_dir)?;
let secret_key = load_or_create_secret(&app_data_dir).await?;
let repository = Repository::open(&app_data_dir).await?;
let store_root = app_data_dir.join("blobs");
let store = FsStore::load(&store_root).await?;
let endpoint = Endpoint::builder(presets::N0)
@@ -341,10 +188,13 @@ impl CoreInner {
endpoint,
router,
store,
repository,
access_policy: AccessPolicy::new(),
event_sink,
active_transfers: TokioMutex::new(HashMap::new()),
active_shares: TokioMutex::new(HashMap::new()),
hash_to_transfer: TokioMutex::new(HashMap::new()),
connection_endpoints: TokioMutex::new(HashMap::new()),
sequence: Mutex::new(1),
});
@@ -387,7 +237,8 @@ impl CoreInner {
json!({ "source_count": sources.len() }),
);
let import = self.import_sources(metadata.transfer_id, sources).await?;
let blob_ticket = BlobTicket::new(self.endpoint.addr(), import.root_hash, BlobFormat::HashSeq);
let blob_ticket =
BlobTicket::new(self.endpoint.addr(), import.root_hash, BlobFormat::HashSeq);
let transfer_name = metadata
.transfer_name
.and_then(non_empty)
@@ -408,10 +259,25 @@ impl CoreInner {
.lock()
.await
.insert(import.root_hash.to_string(), metadata.transfer_id);
self.access_policy
.set_mode(metadata.transfer_id, TransferAccessMode::Public)
.await;
self.active_shares
.lock()
.await
.insert(metadata.transfer_id, import.tag);
self.repository
.upsert_transfer(
metadata.transfer_id,
"send",
"sharing",
Some(&transfer_name),
Some(&import.root_hash.to_string()),
Some(&ticket),
import.file_count,
import.total_size,
)
.await?;
self.emit_transfer(
metadata.transfer_id,
@@ -482,6 +348,32 @@ impl CoreInner {
"receiver_name": receiver_name,
}),
);
self.repository
.upsert_transfer(
transfer_id,
"receive",
"receiving",
parsed
.metadata
.as_ref()
.map(|metadata| metadata.transfer_name.as_str()),
parsed
.metadata
.as_ref()
.map(|metadata| metadata.content_hash.as_str()),
None,
parsed
.metadata
.as_ref()
.map(|metadata| metadata.file_count)
.unwrap_or_default(),
parsed
.metadata
.as_ref()
.map(|metadata| metadata.total_size)
.unwrap_or_default(),
)
.await?;
tokio::fs::create_dir_all(&output_dir).await?;
self.emit_transfer(transfer_id, "receive", "network", "connecting", json!({}));
@@ -527,6 +419,9 @@ impl CoreInner {
let collection = Collection::load(hash_and_format.hash, self.store.as_ref()).await?;
self.export_collection(transfer_id, total_files, output_dir, collection)
.await?;
self.repository
.update_transfer_status(transfer_id, "done")
.await?;
self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({}));
Ok(())
}
@@ -541,19 +436,66 @@ impl CoreInner {
"cancel-requested",
json!({}),
);
self.repository
.update_transfer_status(transfer_id, "cancelled")
.await?;
return Ok(());
}
if self.active_shares.lock().await.remove(&transfer_id).is_some() {
if self
.active_shares
.lock()
.await
.remove(&transfer_id)
.is_some()
{
self.hash_to_transfer
.lock()
.await
.retain(|_, id| *id != transfer_id);
self.access_policy.remove_transfer(transfer_id).await;
self.repository
.update_transfer_status(transfer_id, "stopped")
.await?;
self.emit_transfer(transfer_id, "send", "lifecycle", "share-stopped", json!({}));
return Ok(());
}
anyhow::bail!("transfer not found")
}
async fn set_transfer_access_mode(
&self,
transfer_id: u64,
mode: TransferAccessMode,
) -> Result<()> {
self.access_policy.set_mode(transfer_id, mode.clone()).await;
self.emit_transfer(
transfer_id,
"send",
"access",
"mode-updated",
json!({ "mode": format!("{mode:?}") }),
);
Ok(())
}
async fn approve_endpoint_for_transfer(
&self,
transfer_id: u64,
endpoint_id: String,
) -> Result<()> {
self.access_policy
.approve_endpoint(transfer_id, endpoint_id.clone())
.await;
self.emit_transfer(
transfer_id,
"send",
"access",
"endpoint-approved",
json!({ "endpoint_id": endpoint_id }),
);
Ok(())
}
async fn shutdown(&self) {
self.emit_endpoint("shutdown", "service-shutdown", json!({}));
if let Err(error) = self.router.shutdown().await {
@@ -582,7 +524,11 @@ impl CoreInner {
let stream = read_stream_from_blocking_reader(reader);
let import = core.store.add_stream(stream).await;
let (tag, size) = core
.consume_add_progress(transfer_id, file.collection_name.clone(), import.stream().await)
.consume_add_progress(
transfer_id,
file.collection_name.clone(),
import.stream().await,
)
.await?;
Result::<_>::Ok((file.collection_name, tag, size))
}
@@ -714,7 +660,9 @@ impl CoreInner {
ExportRangesItem::Size(size) => file_size = size,
ExportRangesItem::Data(leaf) => {
if leaf.offset != exported {
anyhow::bail!("export stream for {relative_path} yielded out-of-order data");
anyhow::bail!(
"export stream for {relative_path} yielded out-of-order data"
);
}
exported += leaf.data.len() as u64;
tx.send(Ok(Some(leaf.data)))
@@ -767,6 +715,12 @@ impl CoreInner {
"endpoint_id": message.inner.endpoint_id.map(|id| id.to_string()),
}),
);
if let Some(endpoint_id) = message.inner.endpoint_id {
self.connection_endpoints
.lock()
.await
.insert(message.inner.connection_id, endpoint_id.to_string());
}
let _ = message.tx.send(Ok(())).await;
}
ProviderMessage::ClientConnectedNotify(message) => {
@@ -778,10 +732,48 @@ impl CoreInner {
"endpoint_id": message.inner.endpoint_id.map(|id| id.to_string()),
}),
);
if let Some(endpoint_id) = message.inner.endpoint_id {
self.connection_endpoints
.lock()
.await
.insert(message.inner.connection_id, endpoint_id.to_string());
}
}
ProviderMessage::ConnectionClosed(message) => {
self.connection_endpoints
.lock()
.await
.remove(&message.inner.connection_id);
self.emit_endpoint(
"provider",
"connection-closed",
json!({ "connection_id": message.inner.connection_id }),
);
}
ProviderMessage::GetRequestReceived(message) => {
let transfer_id = self.transfer_for_hash(message.inner.request.hash).await;
if let Some(transfer_id) = transfer_id {
let decision = self
.access_decision(transfer_id, message.inner.connection_id)
.await;
if let AccessDecision::Deny { reason } = decision {
self.emit_transfer(
transfer_id,
"send",
"access",
"request-denied",
json!({
"connection_id": message.inner.connection_id,
"request_id": message.inner.request_id,
"reason": reason,
}),
);
let _ = message
.tx
.send(Err(iroh_blobs::provider::events::AbortReason::Permission))
.await;
return;
}
self.track_request_updates(
transfer_id,
message.inner.connection_id,
@@ -792,7 +784,8 @@ impl CoreInner {
let _ = message.tx.send(Ok(())).await;
}
ProviderMessage::GetRequestReceivedNotify(message) => {
if let Some(transfer_id) = self.transfer_for_hash(message.inner.request.hash).await {
if let Some(transfer_id) = self.transfer_for_hash(message.inner.request.hash).await
{
self.track_request_updates(
transfer_id,
message.inner.connection_id,
@@ -802,8 +795,31 @@ impl CoreInner {
}
}
ProviderMessage::GetManyRequestReceived(message) => {
let transfer_id = self.transfer_for_any_hash(&message.inner.request.hashes).await;
let transfer_id = self
.transfer_for_any_hash(&message.inner.request.hashes)
.await;
if let Some(transfer_id) = transfer_id {
let decision = self
.access_decision(transfer_id, message.inner.connection_id)
.await;
if let AccessDecision::Deny { reason } = decision {
self.emit_transfer(
transfer_id,
"send",
"access",
"request-denied",
json!({
"connection_id": message.inner.connection_id,
"request_id": message.inner.request_id,
"reason": reason,
}),
);
let _ = message
.tx
.send(Err(iroh_blobs::provider::events::AbortReason::Permission))
.await;
return;
}
self.track_request_updates(
transfer_id,
message.inner.connection_id,
@@ -814,7 +830,10 @@ impl CoreInner {
let _ = message.tx.send(Ok(())).await;
}
ProviderMessage::GetManyRequestReceivedNotify(message) => {
if let Some(transfer_id) = self.transfer_for_any_hash(&message.inner.request.hashes).await {
if let Some(transfer_id) = self
.transfer_for_any_hash(&message.inner.request.hashes)
.await
{
self.track_request_updates(
transfer_id,
message.inner.connection_id,
@@ -881,6 +900,18 @@ impl CoreInner {
.find_map(|hash| map.get(&hash.to_string()).copied())
}
async fn access_decision(&self, transfer_id: u64, connection_id: u64) -> AccessDecision {
let endpoint_id = self
.connection_endpoints
.lock()
.await
.get(&connection_id)
.cloned();
self.access_policy
.decide(transfer_id, endpoint_id.as_deref())
.await
}
fn track_request_updates(
self: &Arc<Self>,
transfer_id: u64,
@@ -971,7 +1002,7 @@ impl CoreInner {
let id = format!("{timestamp}-{}", *sequence);
*sequence += 1;
drop(sequence);
self.event_sink.on_event(CoreEvent {
let event = CoreEvent {
id,
timestamp,
scope: scope.to_string(),
@@ -980,308 +1011,14 @@ impl CoreInner {
phase: phase.to_string(),
kind: kind.to_string(),
data_json: data.to_string(),
});
}
}
fn collect_import_files(sources: Vec<ShareSource>) -> Result<Vec<ImportSourceFile>> {
let mut files = Vec::new();
for source in sources {
match source.kind {
SourceKind::Path | SourceKind::IosSecurityScopedUrl => {
let path = source_path(&source)?;
let display_name = source
.display_name
.clone()
.and_then(non_empty)
.or_else(|| path.file_name().and_then(|name| name.to_str()).map(ToOwned::to_owned))
.unwrap_or_else(|| "transfer".to_string());
if source.is_directory || path.is_dir() {
collect_dir_files(&path, &display_name, &mut files)?;
} else {
files.push(ImportSourceFile {
path,
collection_name: validated_relative_string(&display_name)?,
});
}
}
SourceKind::AndroidContentUri => {
anyhow::bail!(
"Android content URI streaming needs platform file descriptor glue before it can be imported without copying"
);
}
}
}
if files.is_empty() {
anyhow::bail!("no files found in selected sources");
}
Ok(files)
}
fn source_path(source: &ShareSource) -> Result<PathBuf> {
if matches!(source.kind, SourceKind::IosSecurityScopedUrl) && source.value.starts_with("file://") {
let without_scheme = source.value.trim_start_matches("file://");
return Ok(PathBuf::from(percent_decode_file_url_path(without_scheme)?));
}
Ok(PathBuf::from(&source.value))
}
fn collect_dir_files(root: &Path, display_name: &str, files: &mut Vec<ImportSourceFile>) -> Result<()> {
for entry in walkdir::WalkDir::new(root).follow_links(false) {
let entry = entry?;
if !entry.file_type().is_file() {
continue;
}
let relative = entry
.path()
.strip_prefix(root)
.context("failed to compute relative path")?;
let collection_name = path_to_string(Path::new(display_name).join(relative), true)?;
files.push(ImportSourceFile {
path: entry.path().to_path_buf(),
collection_name,
});
}
Ok(())
}
fn default_collection_name(files: &[ImportSourceFile]) -> String {
files
.first()
.and_then(|file| file.collection_name.split('/').next())
.filter(|name| !name.is_empty())
.unwrap_or("transfer")
.to_string()
}
fn safe_output_path(output_dir: &Path, relative_path: &str) -> Result<PathBuf> {
let relative = Path::new(relative_path);
path_to_string(relative, true)?;
Ok(output_dir.join(relative))
}
fn read_stream_from_blocking_reader<R>(
mut reader: R,
) -> impl futures::Stream<Item = io::Result<Bytes>> + Send + Sync + 'static
where
R: Read + Send + 'static,
{
let (tx, rx) = async_channel::bounded(2);
std::thread::spawn(move || {
let mut buffer = vec![0; STREAM_BUFFER_LEN];
loop {
match reader.read(&mut buffer) {
Ok(0) => break,
Ok(read) => {
if tx
.send_blocking(Ok(Bytes::copy_from_slice(&buffer[..read])))
.is_err()
{
break;
}
}
Err(error) => {
let _ = tx.send_blocking(Err(error));
break;
}
}
}
});
rx
}
fn write_stream_to_blocking_writer<W>(
mut writer: W,
rx: async_channel::Receiver<io::Result<Option<Bytes>>>,
) -> io::Result<()>
where
W: Write,
{
while let Ok(item) = rx.recv_blocking() {
match item? {
Some(bytes) => writer.write_all(&bytes)?,
None => break,
}
}
writer.flush()
}
async fn wait_for_writer(task: std::thread::JoinHandle<io::Result<()>>) -> Result<io::Result<()>> {
tokio::task::spawn_blocking(move || {
task.join()
.map_err(|_| anyhow::anyhow!("export writer thread panicked"))
})
.await?
}
fn validated_relative_string(name: &str) -> Result<String> {
path_to_string(Path::new(name), true)
}
fn path_to_string(path: impl AsRef<Path>, must_be_relative: bool) -> Result<String> {
let mut path_str = String::new();
let parts = path
.as_ref()
.components()
.filter_map(|component| match component {
Component::Normal(x) => {
let Some(component) = x.to_str() else {
return Some(Err(anyhow::anyhow!("invalid character in path")));
};
if !component.contains('/') && !component.contains('\\') {
Some(Ok(component))
} else {
Some(Err(anyhow::anyhow!("invalid path component {component:?}")))
}
}
Component::RootDir => {
if must_be_relative {
Some(Err(anyhow::anyhow!("invalid root path component")))
} else {
path_str.push('/');
None
}
}
other => Some(Err(anyhow::anyhow!("invalid path component {other:?}"))),
})
.collect::<Result<Vec<_>>>()?;
path_str.push_str(&parts.join("/"));
Ok(path_str)
}
fn percent_decode_file_url_path(value: &str) -> Result<String> {
let bytes = value.as_bytes();
let mut output = Vec::with_capacity(bytes.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'%' {
if i + 2 >= bytes.len() {
anyhow::bail!("invalid percent escape in file URL");
}
let hex = std::str::from_utf8(&bytes[i + 1..i + 3])?;
output.push(u8::from_str_radix(hex, 16).context("invalid percent escape in file URL")?);
i += 3;
} else {
output.push(bytes[i]);
i += 1;
}
}
Ok(String::from_utf8(output)?)
}
async fn load_or_create_secret(app_data_dir: &Path) -> Result<SecretKey> {
if let Ok(secret) = std::env::var("IROH_SECRET") {
return SecretKey::from_str(&secret).context("invalid IROH_SECRET");
}
let path = app_data_dir.join("iroh.secret");
match tokio::fs::read_to_string(&path).await {
Ok(secret) => {
let bytes = HEXLOWER
.decode(secret.trim().as_bytes())
.context("invalid persisted iroh secret encoding")?;
let bytes: [u8; 32] = bytes
.try_into()
.map_err(|_| anyhow::anyhow!("invalid persisted iroh secret length"))?;
Ok(SecretKey::from_bytes(&bytes))
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
let secret = SecretKey::generate();
tokio::fs::write(&path, HEXLOWER.encode(&secret.to_bytes())).await?;
Ok(secret)
}
Err(error) => Err(error.into()),
}
}
fn non_empty(value: String) -> Option<String> {
let trimmed = value.trim();
(!trimmed.is_empty()).then(|| trimmed.to_string())
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or_default()
}
fn unique_transfer_id() -> u64 {
now_ms() as u64
}
uniffi::setup_scaffolding!();
#[cfg(test)]
mod tests {
use super::*;
struct TestSink;
impl CoreEventSink for TestSink {
fn on_event(&self, _event: CoreEvent) {}
}
#[test]
fn metadata_ticket_round_trips() {
let secret = SecretKey::generate();
let addr = iroh::EndpointAddr::new(secret.public());
let blob_ticket = BlobTicket::new(addr, Hash::new([7; 32]), BlobFormat::HashSeq);
let metadata = TransferMetadata::new(
42,
"Summer photos",
Some("hammed".to_string()),
blob_ticket.hash(),
3,
2048,
);
let encoded = VnidropTicket::new(blob_ticket.clone(), metadata.clone())
.encode()
.unwrap();
let parsed = parse_transfer_ticket(&encoded).unwrap();
assert_eq!(parsed.blob_ticket.hash(), blob_ticket.hash());
assert_eq!(parsed.metadata.unwrap().transfer_name, metadata.transfer_name);
}
#[test]
fn invalid_ticket_is_rejected() {
assert!(parse_transfer_ticket("not-a-ticket").is_err());
}
#[tokio::test]
async fn secret_persists() {
let temp = tempfile::tempdir().unwrap();
let first = load_or_create_secret(temp.path()).await.unwrap();
let second = load_or_create_secret(temp.path()).await.unwrap();
assert_eq!(first.to_bytes(), second.to_bytes());
}
#[test]
fn path_validation_rejects_unsafe_paths() {
assert!(path_to_string(Path::new("../escape"), true).is_err());
assert!(path_to_string(Path::new("/absolute"), true).is_err());
assert!(validated_relative_string("bad\\name").is_err());
}
#[test]
fn file_url_decodes_spaces() {
assert_eq!(
percent_decode_file_url_path("/tmp/My%20File.txt").unwrap(),
"/tmp/My File.txt"
);
}
#[test]
fn can_initialize_core() {
let temp = tempfile::tempdir().unwrap();
let core = VnidropCore::initialize(
temp.path().to_string_lossy().to_string(),
Arc::new(TestSink),
)
.unwrap();
let status = core.status();
assert!(!status.endpoint_id.is_empty());
core.shutdown();
let repository = self.repository.clone();
let event_for_repository = event.clone();
tokio::spawn(async move {
if let Err(error) = repository.insert_event(&event_for_repository).await {
tracing::warn!(%error, "failed to persist core event");
}
});
self.event_sink.on_event(event);
}
}

30
crates/vnidrop/src/secret.rs Normal file
View File

@@ -0,0 +1,30 @@
use std::{io, path::Path, str::FromStr};
use anyhow::{Context, Result};
use data_encoding::HEXLOWER;
use iroh::SecretKey;
pub(crate) async fn load_or_create_secret(app_data_dir: &Path) -> Result<SecretKey> {
if let Ok(secret) = std::env::var("IROH_SECRET") {
return SecretKey::from_str(&secret).context("invalid IROH_SECRET");
}
let path = app_data_dir.join("iroh.secret");
match tokio::fs::read_to_string(&path).await {
Ok(secret) => {
let bytes = HEXLOWER
.decode(secret.trim().as_bytes())
.context("invalid persisted iroh secret encoding")?;
let bytes: [u8; 32] = bytes
.try_into()
.map_err(|_| anyhow::anyhow!("invalid persisted iroh secret length"))?;
Ok(SecretKey::from_bytes(&bytes))
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
let secret = SecretKey::generate();
tokio::fs::write(&path, HEXLOWER.encode(&secret.to_bytes())).await?;
Ok(secret)
}
Err(error) => Err(error.into()),
}
}

151
crates/vnidrop/src/tests.rs Normal file
View File

@@ -0,0 +1,151 @@
#[cfg(test)]
mod tests {
use std::{path::Path, sync::Arc};
use iroh::SecretKey;
use iroh_blobs::{ticket::BlobTicket, BlobFormat, Hash};
use crate::{
access_policy::{AccessDecision, AccessPolicy},
api::{CoreEvent, CoreEventSink, TransferMetadata},
filesystem::{path_to_string, percent_decode_file_url_path, validated_relative_string},
repository::Repository,
runtime::VnidropCore,
secret::load_or_create_secret,
ticket::{parse_transfer_ticket, VnidropTicket},
TransferAccessMode,
};
struct TestSink;
impl CoreEventSink for TestSink {
fn on_event(&self, _event: CoreEvent) {}
}
#[test]
fn metadata_ticket_round_trips() {
let secret = SecretKey::generate();
let addr = iroh::EndpointAddr::new(secret.public());
let blob_ticket = BlobTicket::new(addr, Hash::new([7; 32]), BlobFormat::HashSeq);
let metadata = TransferMetadata::new(
42,
"Summer photos",
Some("hammed".to_string()),
blob_ticket.hash(),
3,
2048,
);
let encoded = VnidropTicket::new(blob_ticket.clone(), metadata.clone())
.encode()
.unwrap();
let parsed = parse_transfer_ticket(&encoded).unwrap();
assert_eq!(parsed.blob_ticket.hash(), blob_ticket.hash());
assert_eq!(
parsed.metadata.unwrap().transfer_name,
metadata.transfer_name
);
}
#[test]
fn invalid_ticket_is_rejected() {
assert!(parse_transfer_ticket("not-a-ticket").is_err());
}
#[tokio::test]
async fn secret_persists() {
let temp = tempfile::tempdir().unwrap();
let first = load_or_create_secret(temp.path()).await.unwrap();
let second = load_or_create_secret(temp.path()).await.unwrap();
assert_eq!(first.to_bytes(), second.to_bytes());
}
#[test]
fn path_validation_rejects_unsafe_paths() {
assert!(path_to_string(Path::new("../escape"), true).is_err());
assert!(path_to_string(Path::new("/absolute"), true).is_err());
assert!(validated_relative_string("bad\\name").is_err());
}
#[test]
fn file_url_decodes_spaces() {
assert_eq!(
percent_decode_file_url_path("/tmp/My%20File.txt").unwrap(),
"/tmp/My File.txt"
);
}
#[test]
fn can_initialize_core() {
let temp = tempfile::tempdir().unwrap();
let core = VnidropCore::initialize(
temp.path().to_string_lossy().to_string(),
Arc::new(TestSink),
)
.unwrap();
let status = core.status();
assert!(!status.endpoint_id.is_empty());
core.shutdown();
}
#[tokio::test]
async fn repository_persists_transfers_and_events() {
let temp = tempfile::tempdir().unwrap();
let repository = Repository::open(temp.path()).await.unwrap();
repository
.upsert_transfer(
7,
"send",
"sharing",
Some("demo"),
Some("hash"),
Some("ticket"),
1,
12,
)
.await
.unwrap();
repository
.insert_event(&CoreEvent {
id: "event-1".to_string(),
timestamp: 10,
scope: "transfer".to_string(),
transfer_id: Some(7),
direction: Some("send".to_string()),
phase: "ticket".to_string(),
kind: "created".to_string(),
data_json: "{}".to_string(),
})
.await
.unwrap();
let transfers = repository.list_transfers().await.unwrap();
assert_eq!(transfers.len(), 1);
assert_eq!(transfers[0].transfer_name.as_deref(), Some("demo"));
let events = repository.list_events(Some(7)).await.unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].kind, "created");
}
#[tokio::test]
async fn access_policy_requires_approved_endpoint_when_locked() {
let policy = AccessPolicy::new();
policy
.set_mode(99, TransferAccessMode::ApprovalRequired)
.await;
assert_eq!(
policy.decide(99, Some("node-a")).await,
AccessDecision::Deny {
reason: "approval-required"
}
);
policy.approve_endpoint(99, "node-a".to_string()).await;
assert_eq!(
policy.decide(99, Some("node-a")).await,
AccessDecision::Allow
);
}
}

69
crates/vnidrop/src/ticket.rs Normal file
View File

@@ -0,0 +1,69 @@
use std::str::FromStr;
use anyhow::{Context, Result};
use data_encoding::BASE64URL_NOPAD;
use iroh_blobs::ticket::BlobTicket;
use serde::{Deserialize, Serialize};
use crate::api::TransferMetadata;
const VNIDROP_TICKET_PREFIX: &str = "vnd1:";
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct VnidropTicket {
version: u8,
blob_ticket: String,
metadata: TransferMetadata,
}
impl VnidropTicket {
pub(crate) fn new(blob_ticket: BlobTicket, metadata: TransferMetadata) -> Self {
Self {
version: 1,
blob_ticket: blob_ticket.to_string(),
metadata,
}
}
pub(crate) fn encode(&self) -> Result<String> {
let bytes = serde_json::to_vec(self)?;
Ok(format!(
"{VNIDROP_TICKET_PREFIX}{}",
BASE64URL_NOPAD.encode(&bytes)
))
}
fn decode(value: &str) -> Result<Self> {
let encoded = value
.strip_prefix(VNIDROP_TICKET_PREFIX)
.context("not a VniDrop ticket")?;
let bytes = BASE64URL_NOPAD
.decode(encoded.as_bytes())
.context("invalid VniDrop ticket encoding")?;
serde_json::from_slice(&bytes).context("invalid VniDrop ticket payload")
}
}
#[derive(Debug, Clone)]
pub(crate) struct ParsedTransferTicket {
pub(crate) blob_ticket: BlobTicket,
pub(crate) metadata: Option<TransferMetadata>,
}
pub(crate) fn parse_transfer_ticket(value: &str) -> Result<ParsedTransferTicket> {
if value.starts_with(VNIDROP_TICKET_PREFIX) {
let ticket = VnidropTicket::decode(value)?;
let blob_ticket = BlobTicket::from_str(&ticket.blob_ticket)
.context("invalid BlobTicket inside VniDrop ticket")?;
return Ok(ParsedTransferTicket {
blob_ticket,
metadata: Some(ticket.metadata),
});
}
let blob_ticket = BlobTicket::from_str(value).context("invalid BlobTicket")?;
Ok(ParsedTransferTicket {
blob_ticket,
metadata: None,
})
}

17
crates/vnidrop/src/util.rs Normal file
View File

@@ -0,0 +1,17 @@
use std::time::{SystemTime, UNIX_EPOCH};
pub(crate) fn non_empty(value: String) -> Option<String> {
let trimmed = value.trim();
(!trimmed.is_empty()).then(|| trimmed.to_string())
}
pub(crate) fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or_default()
}
pub(crate) fn unique_transfer_id() -> u64 {
now_ms() as u64
}

10
crates/vnidrop-core/tests/local_transfer.rs → crates/vnidrop/tests/local_transfer.rs
View File

@@ -1,8 +1,6 @@
use std::sync::{Arc, Mutex};
use vnidrop_core::{
CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, VnidropCore,
};
use vnidrop::{CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, VnidropCore};
#[derive(Default)]
struct RecordingSink {
@@ -29,7 +27,11 @@ fn two_local_cores_transfer_file() {
)
.unwrap();
let receiver = VnidropCore::initialize(
receiver_dir.path().join("core").to_string_lossy().to_string(),
receiver_dir
.path()
.join("core")
.to_string_lossy()
.to_string(),
Arc::new(RecordingSink::default()),
)
.unwrap();

0
crates/vnidrop-core/uniffi.toml → crates/vnidrop/uniffi.toml
View File

4
shared/build.gradle.kts
View File

@@ -72,14 +72,14 @@ android {
}
cargo {
packageDirectory = layout.projectDirectory.dir("../crates/vnidrop-core")
packageDirectory = layout.projectDirectory.dir("../crates/vnidrop")
publishJvmArtifacts = true
androidTargetsToBuild.set(setOf(RustAndroidTarget.Arm64))
}
uniffi {
generateFromLibrary {
namespace = "vnidrop_core"
namespace = "vnidrop"
}
}

16
shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt
View File

@@ -8,14 +8,14 @@ import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.withContext
import kotlin.random.Random
import uniffi.vnidrop_core.CoreEvent
import uniffi.vnidrop_core.CoreEventSink
import uniffi.vnidrop_core.ShareMetadataInput
import uniffi.vnidrop_core.ShareResult
import uniffi.vnidrop_core.ShareSource
import uniffi.vnidrop_core.SourceKind
import uniffi.vnidrop_core.TicketInspection
import uniffi.vnidrop_core.VnidropCore
import uniffi.vnidrop.CoreEvent
import uniffi.vnidrop.CoreEventSink
import uniffi.vnidrop.ShareMetadataInput
import uniffi.vnidrop.ShareResult
import uniffi.vnidrop.ShareSource
import uniffi.vnidrop.SourceKind
import uniffi.vnidrop.TicketInspection
import uniffi.vnidrop.VnidropCore
data class CoreUiState(
val isInitialized: Boolean = false,

6
shared/src/iosTest/kotlin/com/vnidrop/app/SharedLogicIOSTest.kt
View File

@@ -3,9 +3,9 @@ package com.vnidrop.app
import platform.Foundation.NSTemporaryDirectory
import kotlin.test.Test
import kotlin.test.assertTrue
import uniffi.vnidrop_core.CoreEvent
import uniffi.vnidrop_core.CoreEventSink
import uniffi.vnidrop_core.VnidropCore
import uniffi.vnidrop.CoreEvent
import uniffi.vnidrop.CoreEventSink
import uniffi.vnidrop.VnidropCore
class SharedLogicIOSTest {

8
shared/src/jvmTest/kotlin/com/vnidrop/app/CoreNativeLoadTest.kt
View File

@@ -3,14 +3,14 @@ package com.vnidrop.app
import java.nio.file.Files
import kotlin.test.Test
import kotlin.test.assertTrue
import uniffi.vnidrop_core.CoreEvent
import uniffi.vnidrop_core.CoreEventSink
import uniffi.vnidrop_core.VnidropCore
import uniffi.vnidrop.CoreEvent
import uniffi.vnidrop.CoreEventSink
import uniffi.vnidrop.VnidropCore
class CoreNativeLoadTest {
@Test
fun generatedBindingsCanInitializeRustCore() {
val coreDir = Files.createTempDirectory("vnidrop-core-jvm-test")
val coreDir = Files.createTempDirectory("vnidrop-jvm-test")
val core = VnidropCore.initialize(
appDataDir = coreDir.toString(),
eventSink = object : CoreEventSink {