Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6ca859c691 | ||
|
|
c2b9167e1e | ||
|
|
4d9985c9cb | ||
|
|
2063d11890 | ||
|
|
0006a188e3 | ||
|
|
687af8bd2a | ||
|
|
c9dbc4a576 | ||
|
|
3382a42b99 | ||
|
|
b7752da694 | ||
|
|
4ff007e403 | ||
|
|
98db87703e |
+3
-10
@@ -12,17 +12,10 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
container: rust:1
|
||||
steps:
|
||||
# Not the checkout action: it runs on node, which the rust:1 job
|
||||
# container doesn't have and this runner doesn't inject (the v5
|
||||
# run died with 'exec: "node": not found'). Plain git needs
|
||||
# neither, and the token URL works against the private repo.
|
||||
# v4 built-in checkout: native Go, needs no Node (rust:1 has none) and
|
||||
# keeps the token out of a fetch URL. Uses the job token for the private repo.
|
||||
- name: checkout
|
||||
run: |
|
||||
git init -q .
|
||||
git fetch --depth 1 \
|
||||
"https://gitea:${{ secrets.GITHUB_TOKEN }}@${GITHUB_SERVER_URL#https://}/${GITHUB_REPOSITORY}.git" \
|
||||
"$GITHUB_SHA"
|
||||
git checkout -q FETCH_HEAD
|
||||
uses: builtin:checkout
|
||||
- name: rustfmt
|
||||
run: |
|
||||
rustup component add rustfmt
|
||||
|
||||
@@ -19,17 +19,10 @@ jobs:
|
||||
matrix:
|
||||
target: [x86_64-unknown-linux-gnu, aarch64-unknown-linux-gnu]
|
||||
steps:
|
||||
# Not the checkout action: it runs on node, which the rust:1 job
|
||||
# container doesn't have and this runner doesn't inject (the v5
|
||||
# run died with 'exec: "node": not found'). Plain git needs
|
||||
# neither, and the token URL works against the private repo.
|
||||
# v4 built-in checkout: native Go, needs no Node (rust:1 has none) and
|
||||
# keeps the token out of a fetch URL. Uses the job token for the private repo.
|
||||
- name: checkout
|
||||
run: |
|
||||
git init -q .
|
||||
git fetch --depth 1 \
|
||||
"https://gitea:${{ secrets.GITHUB_TOKEN }}@${GITHUB_SERVER_URL#https://}/${GITHUB_REPOSITORY}.git" \
|
||||
"$GITHUB_SHA"
|
||||
git checkout -q FETCH_HEAD
|
||||
uses: builtin:checkout
|
||||
|
||||
- name: install tooling
|
||||
run: |
|
||||
@@ -96,3 +89,45 @@ jobs:
|
||||
-H "Authorization: token $TOKEN" \
|
||||
-F "attachment=@$f"
|
||||
done
|
||||
|
||||
# The uhhm Arch registry (pacman.conf: .../api/packages/bl/arch/uhhm),
|
||||
# from the tarballs just released. On klokka: an Arch host with
|
||||
# makepkg. REGISTRY_TOKEN is the uhhm org's write:package token.
|
||||
publish:
|
||||
needs: build
|
||||
runs-on: aarch64
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: builtin:checkout
|
||||
- name: fetch tarballs and pin the PKGBUILD
|
||||
run: |
|
||||
set -eu
|
||||
TAG="${GITHUB_REF_NAME}"
|
||||
VERSION="${TAG#v}"
|
||||
BASE="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/releases/download/${TAG}"
|
||||
cd dist/arch
|
||||
for t in x86_64 aarch64; do
|
||||
f="varde-${VERSION}-${t}-unknown-linux-gnu.tar.gz"
|
||||
# The repo is private: anonymous release downloads 404.
|
||||
curl -fsSL -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" -o "$f" "$BASE/$f"
|
||||
done
|
||||
SUM_X86=$(sha256sum "varde-${VERSION}-x86_64-unknown-linux-gnu.tar.gz" | cut -d' ' -f1)
|
||||
SUM_AARCH=$(sha256sum "varde-${VERSION}-aarch64-unknown-linux-gnu.tar.gz" | cut -d' ' -f1)
|
||||
sed -i "s/^pkgver=.*/pkgver=${VERSION}/" PKGBUILD
|
||||
sed -i "s/sha256sums_x86_64=('.*')/sha256sums_x86_64=('${SUM_X86}')/" PKGBUILD
|
||||
sed -i "s/sha256sums_aarch64=('.*')/sha256sums_aarch64=('${SUM_AARCH}')/" PKGBUILD
|
||||
- name: publish to the uhhm Arch registry
|
||||
run: |
|
||||
set -eu
|
||||
if [ -z "${{ secrets.REGISTRY_TOKEN }}" ]; then
|
||||
echo "::error::REGISTRY_TOKEN not set - add it to publish"
|
||||
exit 1
|
||||
fi
|
||||
cd dist/arch
|
||||
for carch in aarch64 x86_64; do
|
||||
pkgfile=$(CARCH="$carch" makepkg --packagelist | tail -1)
|
||||
CARCH="$carch" makepkg -f --nodeps --noconfirm
|
||||
curl --fail-with-body --user "bl:${{ secrets.REGISTRY_TOKEN }}" \
|
||||
--upload-file "$pkgfile" \
|
||||
"${GITHUB_SERVER_URL}/api/packages/bl/arch/uhhm"
|
||||
done
|
||||
|
||||
@@ -1,2 +1,3 @@
|
||||
target/
|
||||
.claude/
|
||||
.DS_Store
|
||||
|
||||
Generated
+4
-3
@@ -4491,7 +4491,7 @@ checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
|
||||
|
||||
[[package]]
|
||||
name = "varde-ctl"
|
||||
version = "0.1.1"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"clap",
|
||||
@@ -4504,9 +4504,10 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "varde-daemon"
|
||||
version = "0.1.1"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
"iroh",
|
||||
"iroh-blobs",
|
||||
"iroh-mdns-address-lookup",
|
||||
@@ -4529,7 +4530,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "varde-proto"
|
||||
version = "0.1.1"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
|
||||
+1
-1
@@ -3,7 +3,7 @@ resolver = "2"
|
||||
members = ["varde-proto", "varde-daemon", "varde-ctl"]
|
||||
|
||||
[workspace.package]
|
||||
version = "0.1.1"
|
||||
version = "0.1.6"
|
||||
edition = "2021"
|
||||
license = "MIT OR Apache-2.0"
|
||||
repository = "https://project.uhhm.no/uhhm/varde"
|
||||
|
||||
Vendored
+41
@@ -0,0 +1,41 @@
|
||||
# The uhhm Arch registry package, built by .gitea/workflows/release.yml
|
||||
# from the release tarballs (the CI pins pkgver and the sums). For a
|
||||
# from-source build (AUR style) see ../PKGBUILD.
|
||||
pkgname=varde
|
||||
pkgver=0.1.3
|
||||
pkgrel=1
|
||||
pkgdesc="Content-addressed blob mirror daemon on iroh"
|
||||
arch=('x86_64' 'aarch64')
|
||||
url="https://project.uhhm.no/uhhm/varde"
|
||||
license=('MIT OR Apache-2.0')
|
||||
depends=('gcc-libs' 'glibc')
|
||||
optdepends=('networkmanager: metered-connection detection')
|
||||
backup=('etc/varde/config.toml')
|
||||
install=varde.install
|
||||
options=('!strip' '!debug')
|
||||
source_x86_64=("varde-${pkgver}-x86_64-unknown-linux-gnu.tar.gz")
|
||||
source_aarch64=("varde-${pkgver}-aarch64-unknown-linux-gnu.tar.gz")
|
||||
sha256sums_x86_64=('SKIP')
|
||||
sha256sums_aarch64=('SKIP')
|
||||
|
||||
package() {
|
||||
cd "varde-${pkgver}-${CARCH}-unknown-linux-gnu"
|
||||
install -Dm755 bin/varde-daemon "$pkgdir/usr/bin/varde-daemon"
|
||||
install -Dm755 bin/varde-ctl "$pkgdir/usr/bin/varde-ctl"
|
||||
|
||||
install -Dm644 systemd/varde.service "$pkgdir/usr/lib/systemd/system/varde.service"
|
||||
install -Dm644 systemd/varde.socket "$pkgdir/usr/lib/systemd/system/varde.socket"
|
||||
install -Dm644 systemd/user/varde.service "$pkgdir/usr/lib/systemd/user/varde.service"
|
||||
install -Dm644 systemd/user/varde.socket "$pkgdir/usr/lib/systemd/user/varde.socket"
|
||||
# The per-user socket is on for every session, the way PipeWire's
|
||||
# is: it costs nothing until an app connects, and then the daemon
|
||||
# starts by socket activation. The system daemon stays opt-in.
|
||||
install -dm755 "$pkgdir/usr/lib/systemd/user/sockets.target.wants"
|
||||
ln -s ../varde.socket "$pkgdir/usr/lib/systemd/user/sockets.target.wants/varde.socket"
|
||||
|
||||
install -Dm644 config.toml.example "$pkgdir/etc/varde/config.toml"
|
||||
install -Dm644 man/varde.8 "$pkgdir/usr/share/man/man8/varde.8"
|
||||
install -Dm644 man/varde-ctl.1 "$pkgdir/usr/share/man/man1/varde-ctl.1"
|
||||
install -Dm644 LICENSE-MIT "$pkgdir/usr/share/licenses/$pkgname/LICENSE-MIT"
|
||||
install -Dm644 LICENSE-APACHE "$pkgdir/usr/share/licenses/$pkgname/LICENSE-APACHE"
|
||||
}
|
||||
Vendored
+10
@@ -0,0 +1,10 @@
|
||||
post_install() {
|
||||
echo ":: The per-user varde socket starts with your next login."
|
||||
echo " To use it now: systemctl --user daemon-reload && systemctl --user start varde.socket"
|
||||
echo " The system daemon is opt-in: systemctl enable --now varde.socket"
|
||||
}
|
||||
|
||||
post_upgrade() {
|
||||
echo ":: Restart a running daemon to pick up the new version:"
|
||||
echo " systemctl --user try-restart varde.service (and/or: systemctl try-restart varde.service)"
|
||||
}
|
||||
+14
-1
@@ -73,7 +73,8 @@ enum Command {
|
||||
/// Peer trust operations
|
||||
#[command(subcommand)]
|
||||
Peer(PeerCommand),
|
||||
/// Push fully-present content to a trusted peer
|
||||
/// Push fully-present content to a trusted peer, or one allowed to
|
||||
/// exchange pushes
|
||||
Push {
|
||||
/// BLAKE3 hash (hex)
|
||||
hash: String,
|
||||
@@ -116,6 +117,14 @@ enum PeerCommand {
|
||||
/// iroh NodeId (z-base-32)
|
||||
node_id: String,
|
||||
},
|
||||
/// Allow pushes with a peer for a while, without trusting it
|
||||
AllowPush {
|
||||
/// iroh NodeId (z-base-32)
|
||||
node_id: String,
|
||||
/// How long, in seconds (at most a day)
|
||||
#[arg(long, default_value_t = 1800)]
|
||||
ttl: u64,
|
||||
},
|
||||
/// List trusted peers
|
||||
List,
|
||||
}
|
||||
@@ -187,6 +196,9 @@ fn to_request(command: Command) -> Result<Request> {
|
||||
},
|
||||
Command::Peer(PeerCommand::Trust { node_id }) => Request::PeerTrust { node_id },
|
||||
Command::Peer(PeerCommand::Untrust { node_id }) => Request::PeerUntrust { node_id },
|
||||
Command::Peer(PeerCommand::AllowPush { node_id, ttl }) => {
|
||||
Request::PeerAllowPush { node_id, ttl_secs: ttl }
|
||||
}
|
||||
Command::Peer(PeerCommand::List) => Request::PeerList {},
|
||||
Command::Push { hash, node_id } => Request::Push { hash, node_id },
|
||||
Command::Gc => Request::Gc {},
|
||||
@@ -246,6 +258,7 @@ fn print_human(data: Option<&ResponseData>) {
|
||||
match data {
|
||||
ResponseData::Added { hash, bytes } => println!("added {hash} ({bytes} bytes)"),
|
||||
ResponseData::Pinned { hash } => println!("pinned {hash}"),
|
||||
ResponseData::Bytes { data } => println!("{data}"),
|
||||
ResponseData::Done {} => println!("ok"),
|
||||
ResponseData::Materialized { bytes, reflinked } => {
|
||||
let how = if *reflinked { "reflinked" } else { "copied" };
|
||||
|
||||
@@ -18,6 +18,7 @@ fn ctl_status_round_trips() {
|
||||
max_download_bytes_per_sec: 0,
|
||||
discovery: false,
|
||||
wan_upload: false,
|
||||
relay_url: None,
|
||||
gc_interval_secs: 300,
|
||||
};
|
||||
|
||||
|
||||
@@ -16,6 +16,8 @@ required-features = ["socket-api"]
|
||||
varde-proto = { workspace = true }
|
||||
iroh = { workspace = true }
|
||||
iroh-blobs = { workspace = true }
|
||||
# Request::Put / Read carry content inline, base64.
|
||||
base64 = "0.22"
|
||||
# mDNS/local-network address lookup left iroh core in 1.0.
|
||||
iroh-mdns-address-lookup = "0.4"
|
||||
# Interop with iroh-blobs streams and channels; both are already in the
|
||||
|
||||
@@ -24,6 +24,12 @@ pub struct Config {
|
||||
/// Whether any WAN (non-link-local) upload is permitted. Default off:
|
||||
/// LAN-only posture with zero WAN upload.
|
||||
pub wan_upload: bool,
|
||||
/// A self-hosted iroh relay URL for NAT traversal. When set, the
|
||||
/// endpoint uses this relay (`RelayMode::Custom`) instead of n0's
|
||||
/// or none — the sovereign hole-punch coordinator an embedder runs
|
||||
/// on its own infrastructure. Takes precedence over `wan_upload`'s
|
||||
/// n0 defaults. `None` keeps the LAN-only / n0 behaviour.
|
||||
pub relay_url: Option<String>,
|
||||
/// Interval of the store's built-in garbage collector in seconds.
|
||||
/// Since iroh-blobs 0.103 there is no on-demand gc; unpinned blobs
|
||||
/// are swept by this loop.
|
||||
@@ -41,6 +47,7 @@ struct FileConfig {
|
||||
max_download_bytes_per_sec: Option<u64>,
|
||||
discovery: Option<bool>,
|
||||
wan_upload: Option<bool>,
|
||||
relay_url: Option<String>,
|
||||
gc_interval_secs: Option<u64>,
|
||||
}
|
||||
|
||||
@@ -176,6 +183,10 @@ impl Config {
|
||||
wan_upload: env_bool("VARDE_WAN_UPLOAD")?
|
||||
.or(file.wan_upload)
|
||||
.unwrap_or(false),
|
||||
relay_url: std::env::var("VARDE_RELAY_URL")
|
||||
.ok()
|
||||
.filter(|s| !s.is_empty())
|
||||
.or(file.relay_url),
|
||||
gc_interval_secs: env_u64("VARDE_GC_INTERVAL")?
|
||||
.or(file.gc_interval_secs)
|
||||
.unwrap_or(300),
|
||||
|
||||
@@ -335,6 +335,26 @@ impl Daemon {
|
||||
}
|
||||
}
|
||||
|
||||
/// The peer a request names: a NodeId, or a blob ticket from it (whose
|
||||
/// addresses are remembered, so dialing it needs no discovery). An app
|
||||
/// that knows peers by the tickets they hand out - redoal does - can
|
||||
/// allow and push without unpacking them.
|
||||
fn peer_of(&self, peer: &str) -> Result<iroh::EndpointId, OpError> {
|
||||
if let Ok(id) = peer.parse::<iroh::EndpointId>() {
|
||||
return Ok(id);
|
||||
}
|
||||
match crate::transfer::parse_ticket(peer) {
|
||||
Ok((_, _, addr)) => {
|
||||
let id = addr.id;
|
||||
self.transfer.remember(addr);
|
||||
Ok(id)
|
||||
}
|
||||
Err(_) => Err(OpError::InvalidArgument(format!(
|
||||
"{peer:?} is neither a node id nor a ticket"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Trusted peers seen on the LAN recently.
|
||||
fn present_trusted_peers(&self) -> Vec<PeerInfo> {
|
||||
let peers = self.lan_peers.lock().expect("lan peers lock");
|
||||
@@ -371,6 +391,39 @@ impl Daemon {
|
||||
bytes,
|
||||
})
|
||||
}
|
||||
Request::Put { data } => {
|
||||
use base64::Engine as _;
|
||||
let bytes = base64::engine::general_purpose::STANDARD
|
||||
.decode(data.as_bytes())
|
||||
.map_err(|e| OpError::InvalidArgument(format!("put: data is not base64: {e}")))?;
|
||||
if bytes.len() as u64 > varde_proto::MAX_INLINE_BYTES {
|
||||
return Err(OpError::InvalidArgument(format!(
|
||||
"put: {} bytes, more than {} inline; add it by path",
|
||||
bytes.len(),
|
||||
varde_proto::MAX_INLINE_BYTES
|
||||
)));
|
||||
}
|
||||
let (hash, bytes) = self.store.add_bytes(bytes).await?;
|
||||
self.meta.record_format(&hash.to_hex(), StoredFormat::Raw)?;
|
||||
Ok(ResponseData::Added {
|
||||
hash: hash.to_hex().to_string(),
|
||||
bytes,
|
||||
})
|
||||
}
|
||||
Request::Read { hash } => {
|
||||
use base64::Engine as _;
|
||||
let parsed = parse_hash(&hash)?;
|
||||
if self.meta.format_of(&parsed.to_hex()) == StoredFormat::HashSeq {
|
||||
return Err(OpError::InvalidArgument(format!(
|
||||
"read: {} is a collection; materialize it",
|
||||
parsed.to_hex()
|
||||
)));
|
||||
}
|
||||
let bytes = self.store.read_bytes(parsed, varde_proto::MAX_INLINE_BYTES).await?;
|
||||
Ok(ResponseData::Bytes {
|
||||
data: base64::engine::general_purpose::STANDARD.encode(bytes),
|
||||
})
|
||||
}
|
||||
Request::Pin { hash, policy } => {
|
||||
let parsed = parse_hash(&hash)?;
|
||||
self.meta.pin(&parsed.to_hex(), policy)?;
|
||||
@@ -533,24 +586,33 @@ impl Daemon {
|
||||
let parsed: iroh::EndpointId = node_id
|
||||
.parse()
|
||||
.map_err(|e| OpError::InvalidArgument(format!("invalid node id: {e}")))?;
|
||||
if !self.meta.untrust_peer(&parsed.to_string())? {
|
||||
let was_trusted = self.meta.untrust_peer(&parsed.to_string())?;
|
||||
let was_allowed = self.meta.disallow_push(&parsed.to_string())?;
|
||||
if !was_trusted && !was_allowed {
|
||||
return Err(OpError::NotFound(format!("{parsed} was not trusted")));
|
||||
}
|
||||
Ok(ResponseData::Done {})
|
||||
}
|
||||
Request::PeerAllowPush { node_id, ttl_secs } => {
|
||||
let parsed = self.peer_of(&node_id)?;
|
||||
let ttl = ttl_secs.min(varde_proto::MAX_PUSH_TTL_SECS);
|
||||
self.meta
|
||||
.allow_push(&parsed.to_string(), crate::meta::unix_now() + ttl)?;
|
||||
debug!(peer = %parsed, ttl, "pushes allowed");
|
||||
Ok(ResponseData::Done {})
|
||||
}
|
||||
Request::PeerList {} => Ok(ResponseData::Peers {
|
||||
peers: self.present_trusted_peers(),
|
||||
}),
|
||||
Request::Push { hash, node_id } => {
|
||||
let parsed = parse_hash(&hash)?;
|
||||
let peer: iroh::EndpointId = node_id
|
||||
.parse()
|
||||
.map_err(|e| OpError::InvalidArgument(format!("invalid node id: {e}")))?;
|
||||
// Consent, sending side: we only push to peers we trust.
|
||||
// (The receiver independently requires that it trusts us.)
|
||||
if !self.meta.is_trusted(&peer.to_string()) {
|
||||
let peer = self.peer_of(&node_id)?;
|
||||
// Consent, sending side: we only push to peers we trust or
|
||||
// hold a push allowance for. (The receiver independently
|
||||
// requires the same of us.)
|
||||
if !self.meta.may_push(&peer.to_string()) {
|
||||
return Err(OpError::InvalidArgument(format!(
|
||||
"{peer} is not a trusted peer; trust it before pushing"
|
||||
"{peer} is neither trusted nor allowed to exchange pushes; trust it or allow pushes first"
|
||||
)));
|
||||
}
|
||||
let (_, _, complete) = self.store.presence(parsed).await?;
|
||||
|
||||
@@ -40,6 +40,10 @@ struct MetaState {
|
||||
/// Trusted peer NodeIds (z-base-32).
|
||||
#[serde(default)]
|
||||
trusted_peers: BTreeSet<String>,
|
||||
/// Peers allowed to push to us and be pushed to, until a unix time
|
||||
/// (`PeerAllowPush`): consent for that one thing, lapsing on its own.
|
||||
#[serde(default)]
|
||||
push_peers: BTreeMap<String, u64>,
|
||||
/// Hashes (hex) the user shared via ticket export: standing consent
|
||||
/// to serve them to anyone, like an `open_lan` pin.
|
||||
#[serde(default)]
|
||||
@@ -164,6 +168,31 @@ impl Meta {
|
||||
.clone()
|
||||
}
|
||||
|
||||
/// Allow pushes with `node_id` until `until` (unix seconds). Extends,
|
||||
/// never shortens, an allowance already held; expired ones are
|
||||
/// dropped on the way.
|
||||
pub fn allow_push(&self, node_id: &str, until: u64) -> Result<()> {
|
||||
let now = unix_now();
|
||||
self.mutate(|s| {
|
||||
s.push_peers.retain(|_, u| *u > now);
|
||||
let entry = s.push_peers.entry(node_id.to_string()).or_insert(0);
|
||||
*entry = (*entry).max(until);
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether pushes with `node_id` are allowed now: a trusted peer, or
|
||||
/// one holding an unexpired push allowance.
|
||||
pub fn may_push(&self, node_id: &str) -> bool {
|
||||
let state = self.state.lock().expect("meta lock poisoned");
|
||||
state.trusted_peers.contains(node_id)
|
||||
|| state.push_peers.get(node_id).is_some_and(|u| *u > unix_now())
|
||||
}
|
||||
|
||||
/// End a push allowance. Returns whether there was one.
|
||||
pub fn disallow_push(&self, node_id: &str) -> Result<bool> {
|
||||
self.mutate(|s| s.push_peers.remove(node_id).is_some())
|
||||
}
|
||||
|
||||
pub fn is_trusted(&self, node_id: &str) -> bool {
|
||||
self.state
|
||||
.lock()
|
||||
@@ -173,6 +202,14 @@ impl Meta {
|
||||
}
|
||||
}
|
||||
|
||||
/// Seconds since the unix epoch.
|
||||
pub fn unix_now() -> u64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -201,4 +238,27 @@ mod tests {
|
||||
assert!(meta.unpin("h").unwrap());
|
||||
assert!(!meta.is_pinned("h"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_push_allowance_lapses_and_never_shortens() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let meta = Meta::load(dir.path()).unwrap();
|
||||
let now = unix_now();
|
||||
assert!(!meta.may_push("p"));
|
||||
meta.allow_push("p", now + 600).unwrap();
|
||||
assert!(meta.may_push("p"));
|
||||
assert!(!meta.is_trusted("p"), "a push allowance is not trust");
|
||||
meta.allow_push("p", now + 10).unwrap();
|
||||
assert!(meta.may_push("p"), "a shorter ask does not cut it short");
|
||||
meta.allow_push("q", now.saturating_sub(1)).unwrap();
|
||||
assert!(!meta.may_push("q"), "expired");
|
||||
// Survives a reload.
|
||||
let again = Meta::load(dir.path()).unwrap();
|
||||
assert!(again.may_push("p"));
|
||||
assert!(again.disallow_push("p").unwrap());
|
||||
assert!(!again.may_push("p"));
|
||||
// Full trust allows pushes too.
|
||||
again.trust_peer("t").unwrap();
|
||||
assert!(again.may_push("t"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -111,6 +111,8 @@ pub struct Gate {
|
||||
#[derive(Debug, Clone)]
|
||||
struct ConnInfo {
|
||||
trusted: bool,
|
||||
/// Trusted, or holding a push allowance (`PeerAllowPush`).
|
||||
may_push: bool,
|
||||
endpoint_id: Option<String>,
|
||||
}
|
||||
|
||||
@@ -142,6 +144,11 @@ pub fn provider_events(gate: Gate) -> EventSender {
|
||||
.endpoint_id
|
||||
.map(|id| gate.meta.is_trusted(&id.to_string()))
|
||||
.unwrap_or(false);
|
||||
let may_push = m
|
||||
.inner
|
||||
.endpoint_id
|
||||
.map(|id| gate.meta.may_push(&id.to_string()))
|
||||
.unwrap_or(false);
|
||||
let ClientConnected {
|
||||
connection_id,
|
||||
endpoint_id,
|
||||
@@ -151,6 +158,7 @@ pub fn provider_events(gate: Gate) -> EventSender {
|
||||
connection_id,
|
||||
ConnInfo {
|
||||
trusted,
|
||||
may_push,
|
||||
endpoint_id: endpoint_id.map(|id| id.to_string()),
|
||||
},
|
||||
);
|
||||
@@ -192,11 +200,12 @@ pub fn provider_events(gate: Gate) -> EventSender {
|
||||
ProviderMessage::ObserveRequestReceived(m) => {
|
||||
// Presence is information too: gate it like data.
|
||||
// Bitfields are metadata, not payload, so trusted
|
||||
// peers may observe even while serving is off or the
|
||||
// connection is metered (push confirmation needs it).
|
||||
// peers - and peers allowed to push, whose push is
|
||||
// confirmed by observing it - may observe even while
|
||||
// serving is off or the connection is metered.
|
||||
let trusted = conns
|
||||
.get(&m.inner.connection_id)
|
||||
.map(|i| i.trusted)
|
||||
.map(|i| i.trusted || i.may_push)
|
||||
.unwrap_or(false);
|
||||
let allowed = trusted
|
||||
|| allow(&gate, &conns, m.inner.connection_id, [m.inner.request.hash]);
|
||||
@@ -208,12 +217,13 @@ pub fn provider_events(gate: Gate) -> EventSender {
|
||||
m.tx.send(verdict).await.ok();
|
||||
}
|
||||
ProviderMessage::PushRequestReceived(m) => {
|
||||
// Inbound writes: trusted peers only, deferred while
|
||||
// metered (RateLimited invites a retry later).
|
||||
// Inbound writes: trusted peers and push allowances
|
||||
// only, deferred while metered (RateLimited invites a
|
||||
// retry later).
|
||||
let info = conns.get(&m.inner.connection_id);
|
||||
let trusted = info.map(|i| i.trusted).unwrap_or(false);
|
||||
debug!(hash = %m.inner.request.hash.to_hex(), trusted, "push request received");
|
||||
let verdict = if !trusted {
|
||||
let may_push = info.map(|i| i.may_push).unwrap_or(false);
|
||||
debug!(hash = %m.inner.request.hash.to_hex(), may_push, "push request received");
|
||||
let verdict = if !may_push {
|
||||
Err(AbortReason::Permission)
|
||||
} else if gate.metered.is_metered() {
|
||||
Err(AbortReason::RateLimited)
|
||||
|
||||
@@ -228,6 +228,30 @@ impl BlobStore {
|
||||
|
||||
/// Export `hash` to `dest`. Returns bytes written and whether at least
|
||||
/// one file was reflinked.
|
||||
/// Import `data` as one raw blob (`Request::Put`).
|
||||
pub async fn add_bytes(&self, data: Vec<u8>) -> Result<(Hash, u64), OpError> {
|
||||
let size = data.len() as u64;
|
||||
let tag = self.store.add_bytes(data).temp_tag().await.map_err(internal)?;
|
||||
info!(hash = %tag.hash().to_hex(), size, "imported bytes");
|
||||
Ok((tag.hash(), size))
|
||||
}
|
||||
|
||||
/// The whole content of a complete raw blob (`Request::Read`), when it
|
||||
/// is no larger than `limit`.
|
||||
pub async fn read_bytes(&self, hash: Hash, limit: u64) -> Result<Vec<u8>, OpError> {
|
||||
match self.store.blobs().status(hash).await.map_err(internal)? {
|
||||
BlobStatus::Complete { size } if size > limit => Err(OpError::InvalidArgument(format!(
|
||||
"{} is {size} bytes, more than {limit} can be read inline; materialize it",
|
||||
hash.to_hex()
|
||||
))),
|
||||
BlobStatus::Complete { .. } => {
|
||||
let bytes = self.store.blobs().get_bytes(hash).await.map_err(internal)?;
|
||||
Ok(bytes.to_vec())
|
||||
}
|
||||
_ => Err(OpError::NotFound(format!("{} is not complete in the store", hash.to_hex()))),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn materialize(
|
||||
&self,
|
||||
hash: Hash,
|
||||
|
||||
@@ -62,10 +62,19 @@ impl Transfer {
|
||||
metered: MeteredState,
|
||||
) -> Result<Transfer> {
|
||||
let secret = load_or_create_secret(&config.store_dir.join("secret.key"))?;
|
||||
// With wan_upload the n0 defaults apply (their relays and DNS
|
||||
// lookup, matching the pre-1.0 default relay mode); otherwise the
|
||||
// endpoint gets no external services at all.
|
||||
let builder = if config.wan_upload {
|
||||
// A self-hosted relay (relay_url) is the sovereign hole-punch
|
||||
// path and takes precedence: the endpoint coordinates through
|
||||
// the operator's own iroh relay, never n0's. Failing that,
|
||||
// wan_upload opts into n0's relays + DNS; otherwise the endpoint
|
||||
// gets no external services at all (LAN-only).
|
||||
let builder = if let Some(url) = &config.relay_url {
|
||||
let relay_url: iroh::RelayUrl = url
|
||||
.parse()
|
||||
.with_context(|| format!("parsing relay_url {url:?}"))?;
|
||||
let relay_map = iroh::RelayMap::from(relay_url);
|
||||
Endpoint::builder(iroh::endpoint::presets::Minimal)
|
||||
.relay_mode(RelayMode::Custom(relay_map))
|
||||
} else if config.wan_upload {
|
||||
Endpoint::builder(iroh::endpoint::presets::N0)
|
||||
} else {
|
||||
Endpoint::builder(iroh::endpoint::presets::Minimal).relay_mode(RelayMode::Disabled)
|
||||
@@ -289,6 +298,15 @@ impl Transfer {
|
||||
Ok(ticket.to_string())
|
||||
}
|
||||
|
||||
/// Teach the endpoint where a peer can be reached (the addresses a
|
||||
/// ticket carries), so a push dials it directly instead of waiting on
|
||||
/// discovery.
|
||||
pub fn remember(&self, addr: EndpointAddr) {
|
||||
if !addr.addrs.is_empty() {
|
||||
self.known_addrs.add_endpoint_info(addr);
|
||||
}
|
||||
}
|
||||
|
||||
/// Queue a background fetch of `content` from `providers`. Emits
|
||||
/// progress events and [`Event::PinComplete`] when fully verified.
|
||||
pub fn spawn_fetch(&self, content: HashAndFormat, providers: Vec<EndpointAddr>) {
|
||||
|
||||
@@ -422,3 +422,61 @@ fn push_rejected_without_receiver_trust() {
|
||||
// And the receiver holds none of it.
|
||||
assert!(!is_complete(&b.socket, &hash));
|
||||
}
|
||||
|
||||
/// A push allowance is enough on both sides: two daemons that only
|
||||
/// allowed pushes with each other (strangers sharing a place, in redoal)
|
||||
/// exchange content without trusting each other - and the allowance
|
||||
/// grants nothing else, so neither lists the other as a trusted peer.
|
||||
#[test]
|
||||
fn push_allowance_is_enough_and_is_not_trust() {
|
||||
let _serial = MDNS_SERIAL.lock().unwrap_or_else(|e| e.into_inner());
|
||||
let dir_a = tempfile::tempdir().unwrap();
|
||||
let dir_b = tempfile::tempdir().unwrap();
|
||||
let a = support::spawn_daemon_with_env(dir_a.path(), &[("VARDE_DISCOVERY", "true")]);
|
||||
let b = support::spawn_daemon_with_env(dir_b.path(), &[("VARDE_DISCOVERY", "true")]);
|
||||
let mut client_a = support::Client::connect(&a.socket);
|
||||
let mut client_b = support::Client::connect(&b.socket);
|
||||
let id_a = node_id(&mut client_a);
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let src = dir.path().join("voice.bin");
|
||||
let payload = vec![7u8; 120_000];
|
||||
std::fs::write(&src, &payload).unwrap();
|
||||
let hash = add_file(&mut client_a, &src);
|
||||
pin(&mut client_a, &hash, false);
|
||||
|
||||
// A knows B by a ticket B handed out (as redoal peers do), not by
|
||||
// its bare node id.
|
||||
let b_src = dir.path().join("b-own.bin");
|
||||
std::fs::write(&b_src, vec![1u8; 1_000]).unwrap();
|
||||
let b_hash = add_file(&mut client_b, &b_src);
|
||||
let b_ticket = match client_b.request(&Request::TicketExport { hash: b_hash }).data {
|
||||
Some(ResponseData::Ticket { ticket }) => ticket,
|
||||
other => panic!("unexpected ticket reply: {other:?}"),
|
||||
};
|
||||
assert!(client_a.request(&Request::PeerAllowPush { node_id: b_ticket.clone(), ttl_secs: 600 }).ok);
|
||||
assert!(client_b.request(&Request::PeerAllowPush { node_id: id_a, ttl_secs: 600 }).ok);
|
||||
match client_a.request(&Request::PeerList {}).data {
|
||||
Some(ResponseData::Peers { peers }) => assert!(peers.is_empty(), "an allowance is not trust: {peers:?}"),
|
||||
other => panic!("unexpected peer list: {other:?}"),
|
||||
}
|
||||
|
||||
// The peer's address arrives by mdns; retry until it does.
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
let reply = loop {
|
||||
let reply = client_a.request(&Request::Push { hash: hash.clone(), node_id: b_ticket.clone() });
|
||||
if reply.ok || Instant::now() > deadline {
|
||||
break reply;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(500));
|
||||
};
|
||||
if !reply.ok {
|
||||
let message = reply.error.map(|e| e.message).unwrap_or_default();
|
||||
// Only an unreachable peer is the environment's fault; any other
|
||||
// refusal is a real failure.
|
||||
assert!(message.contains("connect"), "push failed: {message}");
|
||||
eprintln!("SKIP: push never reached the peer (multicast unavailable?): {message}");
|
||||
return;
|
||||
}
|
||||
assert!(wait_complete(&b.socket, &hash, Duration::from_secs(60)), "pushed content not complete on receiver");
|
||||
}
|
||||
|
||||
@@ -276,3 +276,45 @@ fn reflink_when_supported() {
|
||||
drop(daemon);
|
||||
let _ = std::fs::remove_dir_all(&base);
|
||||
}
|
||||
|
||||
/// Put and Read carry the bytes on the socket: a client that shares no
|
||||
/// filesystem with the daemon (a desktop app and the system daemon, which
|
||||
/// runs as its own user behind ProtectHome) stores and reads back without
|
||||
/// a path either side must reach. A Put blob reads back by Materialize too.
|
||||
#[test]
|
||||
fn bytes_round_trip_inline() {
|
||||
use base64::Engine as _;
|
||||
let b64 = base64::engine::general_purpose::STANDARD;
|
||||
let daemon = support::spawn_daemon();
|
||||
let mut client = support::Client::connect(&daemon.socket);
|
||||
let payload: Vec<u8> = (0..50_000u32).flat_map(|i| i.to_le_bytes()).collect();
|
||||
|
||||
let reply = client.request(&Request::Put { data: b64.encode(&payload) });
|
||||
assert!(reply.ok, "put failed: {:?}", reply.error);
|
||||
let hash = match reply.data {
|
||||
Some(ResponseData::Added { hash, bytes }) => {
|
||||
assert_eq!(bytes, payload.len() as u64);
|
||||
hash
|
||||
}
|
||||
other => panic!("unexpected put reply: {other:?}"),
|
||||
};
|
||||
|
||||
let reply = client.request(&Request::Read { hash: hash.clone() });
|
||||
assert!(reply.ok, "read failed: {:?}", reply.error);
|
||||
match reply.data {
|
||||
Some(ResponseData::Bytes { data }) => assert_eq!(b64.decode(data).unwrap(), payload),
|
||||
other => panic!("unexpected read reply: {other:?}"),
|
||||
}
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let dest = dir.path().join("out.bin");
|
||||
materialize(&mut client, &hash, &dest, MaterializeMode::Copy).unwrap();
|
||||
assert_eq!(std::fs::read(&dest).unwrap(), payload);
|
||||
|
||||
// Not base64: refused, not stored.
|
||||
let reply = client.request(&Request::Put { data: "not base64!".into() });
|
||||
assert_eq!(reply.error.expect("error body").code, ErrorCode::InvalidArgument);
|
||||
// Unknown hash: not found.
|
||||
let reply = client.request(&Request::Read { hash: "0".repeat(64) });
|
||||
assert!(!reply.ok);
|
||||
}
|
||||
|
||||
@@ -20,6 +20,13 @@ use serde::{Deserialize, Serialize};
|
||||
/// Bumped on incompatible changes; reported in [`GlobalStatus`].
|
||||
pub const PROTOCOL_VERSION: u32 = 2;
|
||||
|
||||
/// The longest a [`Request::PeerAllowPush`] allowance may last: a day.
|
||||
pub const MAX_PUSH_TTL_SECS: u64 = 24 * 60 * 60;
|
||||
|
||||
/// The most bytes a [`Request::Put`] or a [`Request::Read`] reply carries
|
||||
/// inline. Larger content goes through `Add`/`Materialize` and a path.
|
||||
pub const MAX_INLINE_BYTES: u64 = 16 * 1024 * 1024;
|
||||
|
||||
/// A request from a client to the daemon. One JSON object per line.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(tag = "type", rename_all = "snake_case")]
|
||||
@@ -61,6 +68,24 @@ pub enum Request {
|
||||
#[serde(default)]
|
||||
mode: MaterializeMode,
|
||||
},
|
||||
/// Import bytes carried in the request, base64 (standard alphabet),
|
||||
/// at most [`MAX_INLINE_BYTES`]. Replies [`ResponseData::Added`].
|
||||
///
|
||||
/// For a client that shares no filesystem with the daemon: the system
|
||||
/// daemon runs as its own user with `ProtectHome`, so it can neither
|
||||
/// read a file a desktop app wrote for `Add` nor write one back for
|
||||
/// `Materialize`. Added in varde 0.1.4.
|
||||
Put {
|
||||
/// The content, base64.
|
||||
data: String,
|
||||
},
|
||||
/// Read a complete raw blob back in the reply, base64, at most
|
||||
/// [`MAX_INLINE_BYTES`]. Replies [`ResponseData::Bytes`]. The
|
||||
/// counterpart of [`Request::Put`]; added in varde 0.1.4.
|
||||
Read {
|
||||
/// Hash (hex) of the blob.
|
||||
hash: String,
|
||||
},
|
||||
/// Query daemon status, globally or for one hash.
|
||||
Status {
|
||||
/// If set, report on this hash; otherwise report global status.
|
||||
@@ -94,6 +119,20 @@ pub enum Request {
|
||||
/// The peer's iroh NodeId (z-base-32).
|
||||
node_id: String,
|
||||
},
|
||||
/// Allow pushes between this daemon and a peer for `ttl_secs` (at most
|
||||
/// [`MAX_PUSH_TTL_SECS`]): we may push to it, and accept what it pushes
|
||||
/// to us. Nothing else - unlike [`Request::PeerTrust`] it neither
|
||||
/// syncs the peer's pins nor serves it our content - so an app can
|
||||
/// grant it to strangers it shares something with for a while (two
|
||||
/// phones at the same place) and let it lapse. Asking again extends
|
||||
/// it; [`Request::PeerUntrust`] ends it early. Added in varde 0.1.5.
|
||||
PeerAllowPush {
|
||||
/// The peer's iroh NodeId (z-base-32), or a blob ticket it issued
|
||||
/// (varde 0.1.6+), whose addresses are then used to reach it.
|
||||
node_id: String,
|
||||
/// How long the allowance lasts, in seconds.
|
||||
ttl_secs: u64,
|
||||
},
|
||||
/// List trusted peers and their current connectivity.
|
||||
PeerList {},
|
||||
/// Push fully-present content to a trusted peer, unprompted.
|
||||
@@ -104,6 +143,9 @@ pub enum Request {
|
||||
/// policy), so the content survives gc and shows up in `List`.
|
||||
Push {
|
||||
/// Hash (hex) of the fully-present blob or HashSeq root to push.
|
||||
///
|
||||
/// The receiver below may be a NodeId or, from varde 0.1.6, a blob
|
||||
/// ticket it issued.
|
||||
hash: String,
|
||||
/// The receiving peer's iroh NodeId (z-base-32).
|
||||
node_id: String,
|
||||
@@ -228,6 +270,11 @@ pub enum ResponseData {
|
||||
/// Whether at least one file was reflinked rather than copied.
|
||||
reflinked: bool,
|
||||
},
|
||||
/// A blob's content, base64. Reply to [`Request::Read`].
|
||||
Bytes {
|
||||
/// The content, base64 (standard alphabet).
|
||||
data: String,
|
||||
},
|
||||
/// Global daemon status. Reply to [`Request::Status`] without a hash.
|
||||
Status(GlobalStatus),
|
||||
/// Per-hash status. Reply to [`Request::Status`] with a hash.
|
||||
|
||||
Reference in New Issue
Block a user