diff --git a/Cargo.lock b/Cargo.lock index 835dba5..3ace047 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -170,6 +170,18 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "async-broadcast" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" +dependencies = [ + "event-listener", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + [[package]] name = "async-channel" version = "2.5.0" @@ -195,6 +207,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "async-recursion" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -890,6 +913,12 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" +[[package]] +name = "endi" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66b7e2430c6dff6a955451e2cfc438f09cea1965a9d6f87f7e3b90decc014099" + [[package]] name = "enum-as-inner" version = "0.6.1" @@ -909,6 +938,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1027f7680c853e056ebcec683615fb6fbbc07dbaa13b4d5d9442b146ded4ecef" dependencies = [ "enumflags2_derive", + "serde", ] [[package]] @@ -2132,6 +2162,15 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "minimal-lexical" version = "0.2.1" @@ -2519,6 +2558,16 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "ordered-stream" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9aa2b01e1d916879f73a53d01d1d6cee68adbb31d6d9177a8cfce093cced1d50" +dependencies = [ + "futures-core", + "pin-project-lite", +] + [[package]] name = "p256" version = "0.13.2" @@ -3652,6 +3701,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_repr" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "serde_spanned" version = "0.6.9" @@ -4196,6 +4256,7 @@ dependencies = [ "signal-hook-registry", "socket2 0.6.5", "tokio-macros", + "tracing", "windows-sys 0.61.2", ] @@ -4505,6 +4566,17 @@ version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" +[[package]] +name = "uds_windows" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e" +dependencies = [ + "memoffset", + "tempfile", + "windows-sys 0.61.2", +] + [[package]] name = "unarray" version = "0.1.4" @@ -4581,6 +4653,7 @@ checksum = "ea5fab0d6c3c01ae70085a09cb03d4c7a1d6314e2b3e075392783396d724ca0a" dependencies = [ "getrandom 0.4.3", "js-sys", + "serde_core", "wasm-bindgen", ] @@ -4627,6 +4700,7 @@ dependencies = [ "tracing", "tracing-subscriber", "varde-proto", + "zbus", ] [[package]] @@ -5298,6 +5372,62 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2164e798d9e3d84ee2c91139ace54638059a3b23e361f5c11781c2c6459bde0f" +[[package]] +name = "zbus" +version = "5.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a28b97f866896a4be7aefd2b5a8e01bb6773d19a775d54ab28b4d094b9a4480e" +dependencies = [ + "async-broadcast", + "async-recursion", + "async-trait", + "enumflags2", + "event-listener", + "futures-core", + "futures-lite", + "hex", + "libc", + "ordered-stream", + "rustix", + "serde", + "serde_repr", + "tokio", + "tracing", + "uds_windows", + "uuid", + "windows-sys 0.61.2", + "winnow 1.0.4", + "zbus_macros", + "zbus_names", + "zvariant", +] + +[[package]] +name = "zbus_macros" +version = "5.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e05ad887425eecf5e8384dc2406a4a9313eb73468712fc1cdea362eb4fe0469" +dependencies = [ + "proc-macro-crate", + "proc-macro2", + "quote", + "syn 2.0.118", + "zbus_names", + "zvariant", + "zvariant_utils", +] + +[[package]] +name = "zbus_names" +version = "4.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1039ca249fee9559680f3a9f05b55e0761fee51af4f6c1e7d8c1f31e549721d2" +dependencies = [ + "serde", + "winnow 1.0.4", + "zvariant", +] + [[package]] name = "zerocopy" version = "0.8.54" @@ -5383,3 +5513,43 @@ name = "zmij" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" + +[[package]] +name = "zvariant" +version = "5.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cf057bb00bf5c9ad77abb6147b0ca4818236a1858416e9d988e40d6322fefa7" +dependencies = [ + "endi", + "enumflags2", + "serde", + "winnow 1.0.4", + "zvariant_derive", + "zvariant_utils", +] + +[[package]] +name = "zvariant_derive" +version = "5.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8118ca6bda77bfc0ab51d660db0c955f2505eef854c9a449435bccb616933b31" +dependencies = [ + "proc-macro-crate", + "proc-macro2", + "quote", + "syn 2.0.118", + "zvariant_utils", +] + +[[package]] +name = "zvariant_utils" +version = "3.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90cb9383f9b45290407a1258b202d3f8f01db719eb60b4e4055c6375af4fc7c7" +dependencies = [ + "proc-macro2", + "quote", + "serde", + "syn 2.0.118", + "winnow 1.0.4", +] diff --git a/dist/PKGBUILD b/dist/PKGBUILD new file mode 100644 index 0000000..7957e43 --- /dev/null +++ b/dist/PKGBUILD @@ -0,0 +1,49 @@ +# Maintainer: you +# +# UNTESTED SKELETON — best-effort Arch packaging for varde. Review before +# building; in particular pkgver/source must point at a real release. + +pkgname=varde +pkgver=0.1.0 +pkgrel=1 +pkgdesc="Content-addressed blob mirror daemon on iroh" +arch=('x86_64' 'aarch64') +url="https://github.com/bendiklh/varde" +license=('MIT OR Apache-2.0') +depends=('gcc-libs') +optdepends=('networkmanager: metered-connection detection') +makedepends=('cargo' 'scdoc') +source=("$pkgname-$pkgver.tar.gz::$url/archive/v$pkgver.tar.gz") +sha256sums=('SKIP') + +build() { + cd "$pkgname-$pkgver" + export RUSTUP_TOOLCHAIN=stable + export CARGO_TARGET_DIR=target + cargo build --release --locked + scdoc < dist/varde.8.scd > varde.8 + scdoc < dist/varde-ctl.1.scd > varde-ctl.1 +} + +check() { + cd "$pkgname-$pkgver" + # Integration tests spawn daemons and use the network; keep package + # builds to the fast unit tests. + cargo test --release --locked --lib -p varde-proto -p varde-daemon +} + +package() { + cd "$pkgname-$pkgver" + install -Dm755 target/release/varde-daemon "$pkgdir/usr/bin/varde-daemon" + install -Dm755 target/release/varde-ctl "$pkgdir/usr/bin/varde-ctl" + + install -Dm644 dist/varde.service "$pkgdir/usr/lib/systemd/system/varde.service" + install -Dm644 dist/varde.socket "$pkgdir/usr/lib/systemd/system/varde.socket" + install -Dm644 dist/user/varde.service "$pkgdir/usr/lib/systemd/user/varde.service" + install -Dm644 dist/user/varde.socket "$pkgdir/usr/lib/systemd/user/varde.socket" + + install -Dm644 dist/config.toml "$pkgdir/etc/varde/config.toml" + + install -Dm644 varde.8 "$pkgdir/usr/share/man/man8/varde.8" + install -Dm644 varde-ctl.1 "$pkgdir/usr/share/man/man1/varde-ctl.1" +} diff --git a/dist/config.toml b/dist/config.toml new file mode 100644 index 0000000..be1918b --- /dev/null +++ b/dist/config.toml @@ -0,0 +1,28 @@ +# varde example configuration — /etc/varde/config.toml (system) or +# $XDG_CONFIG_HOME/varde/config.toml (user). +# +# Every key is optional; the daemon runs with an empty (or absent) +# config. Values shown are the defaults. Precedence: +# CLI flags > VARDE_* environment > this file > defaults. + +# Directory holding the blob store, metadata and endpoint secret key. +# Default: /var/lib/varde (system) or $XDG_DATA_HOME/varde (user). +#store_dir = "/var/lib/varde" + +# Unix socket serving the JSON-lines API. +# Default: /run/varde/varde.sock (system) or $XDG_RUNTIME_DIR/varde.sock. +#socket_path = "/run/varde/varde.sock" + +# Upload cap in bytes/sec. 0 disables serving entirely. +#max_upload_bytes_per_sec = 10485760 # 10 MiB/s + +# Download cap in bytes/sec. 0 means unlimited. +#max_download_bytes_per_sec = 0 + +# LAN peer discovery (mDNS). Discovery is the open tier: being visible +# costs nothing; serving is still gated on trust and per-pin policy. +#discovery = true + +# Permit WAN traffic (enables the iroh relay for NAT traversal). Off by +# default: LAN-only posture, zero WAN upload. +#wan_upload = false diff --git a/dist/user/varde.service b/dist/user/varde.service new file mode 100644 index 0000000..ee40951 --- /dev/null +++ b/dist/user/varde.service @@ -0,0 +1,15 @@ +# varde per-user service (systemctl --user). Pairs with the user +# varde.socket; stores under $XDG_DATA_HOME/varde. +[Unit] +Description=varde content-addressed blob mirror daemon (user) +Documentation=man:varde(8) +Requires=varde.socket + +[Service] +Type=notify +ExecStart=/usr/bin/varde-daemon --user +Restart=on-failure +NoNewPrivileges=yes + +[Install] +WantedBy=default.target diff --git a/dist/user/varde.socket b/dist/user/varde.socket new file mode 100644 index 0000000..bc0f41d --- /dev/null +++ b/dist/user/varde.socket @@ -0,0 +1,9 @@ +# Socket activation for the per-user varde daemon. +[Unit] +Description=varde API socket (user) + +[Socket] +ListenStream=%t/varde.sock + +[Install] +WantedBy=sockets.target diff --git a/dist/varde-ctl.1.scd b/dist/varde-ctl.1.scd new file mode 100644 index 0000000..31526ca --- /dev/null +++ b/dist/varde-ctl.1.scd @@ -0,0 +1,84 @@ +varde-ctl(1) + +# NAME + +varde-ctl - control the varde blob mirror daemon + +# SYNOPSIS + +*varde-ctl* [*--socket* _PATH_] [*--json*] _COMMAND_ [_ARGS_] + +# DESCRIPTION + +Command-line client for *varde*(8), speaking the JSON Lines socket +protocol. Doubles as the protocol's reference implementation: every +subcommand maps 1:1 onto a request. + +# GLOBAL OPTIONS + +*--socket* _PATH_ + Daemon socket (env: *VARDE_SOCKET*). Default: + _/run/varde/varde.sock_ as root, _$XDG_RUNTIME_DIR/varde.sock_ + otherwise. + +*--json* + Print the raw JSON response envelope instead of human output. + +# COMMANDS + +*add* [-r|--recursive] _PATH_ + Import a file (or, with *-r*, a directory tree) into the store. + Prints the root hash. Content is not pinned by adding — pin it to + protect it from *gc*. + +*pin* [--open-lan] _HASH_ + Standing intent: keep this content, fetch it if absent, serve it to + trusted peers. *--open-lan* serves it to any LAN peer. + +*unpin* _HASH_ + Remove the intent. Data stays until *gc*. + +*materialize* [--mode reflink-or-copy|copy] _HASH_ _DEST_ + Export content to a path. The default mode reflinks when the + filesystem supports it and falls back to copying. + +*status* [_HASH_] + Daemon status, or presence/completeness of one hash. + +*list* + All pins with policy and completeness. + +*ticket export* _HASH_ + Print an iroh ticket for out-of-band sharing. Note: exporting + grants standing consent to serve that content to anyone (the + ticket holder is not on the allowlist). + +*ticket import* [--open-lan] _TICKET_ + Pin the ticket's content and fetch it from the embedded provider. + +*peer trust* _NODE-ID_ + Add a peer to the allowlist. Content is served to and auto-fetched + from trusted peers only. Applies to connections made after the + change. + +*peer untrust* _NODE-ID_ + Remove a peer from the allowlist. + +*peer list* + Trusted peers and whether they are currently present on the LAN. + +*gc* + Drop all blobs not reachable from a pin. Never runs implicitly. + +*subscribe* + Stream daemon events (transfer progress, peer presence, pin + completion) as JSON lines until interrupted. + +# EXIT STATUS + +0 on success; nonzero with a structured error (code and message) on +stderr otherwise. + +# SEE ALSO + +*varde*(8) diff --git a/dist/varde.8.scd b/dist/varde.8.scd new file mode 100644 index 0000000..65fbeb8 --- /dev/null +++ b/dist/varde.8.scd @@ -0,0 +1,77 @@ +varde(8) + +# NAME + +varde-daemon - content-addressed blob mirror daemon on iroh + +# SYNOPSIS + +*varde-daemon* [_OPTIONS_] + +# DESCRIPTION + +varde owns a content-addressed blob store and mirrors content between +consenting peers. Applications talk to it over a unix socket speaking +JSON Lines (one request per line, one response per line); *varde-ctl*(1) +is the reference client. + +Discovery is open; replication is explicit. The daemon advertises itself +via mDNS on the local network (identifiable, never camouflaged), but +content is only served to peers on the trust allowlist, plus content +deliberately published per-pin (*open-lan*) or via ticket export. All +fetched data is BLAKE3-verified regardless of source. + +Default traffic posture: LAN only (no relay), upload capped at 10 MiB/s, +UDP sockets marked DSCP CS1 (best effort), and all transfers stop while +NetworkManager reports the connection as metered. + +# OPTIONS + +*--config* _PATH_ + Config file. Default: _/etc/varde/config.toml_ in system mode, + _$XDG_CONFIG_HOME/varde/config.toml_ in user mode. The default file + may be absent; an explicitly named one must exist. + +*--store* _PATH_ + Store directory override. + +*--socket* _PATH_ + API socket path override. + +*--system*, *--user* + Force system-mode or user-mode default paths. Default: system when + running as root, user otherwise. + +*--version*, *--help* + Print version or usage and exit. + +# CONFIGURATION + +See the commented example at _/etc/varde/config.toml_. Precedence: +flags > *VARDE_\** environment variables (*VARDE_CONFIG*, *VARDE_STORE*, +*VARDE_SOCKET*, *VARDE_MAX_UPLOAD*, *VARDE_MAX_DOWNLOAD*, +*VARDE_DISCOVERY*, *VARDE_WAN_UPLOAD*) > config file > defaults. + +# FILES + +_/var/lib/varde_ (system), _$XDG_DATA_HOME/varde_ (user) + Store directory: iroh-blobs store under _blobs/_, pins and trust in + _meta.json_, endpoint key in _secret.key_ (mode 0600). + +_/run/varde/varde.sock_ (system), _$XDG_RUNTIME_DIR/varde.sock_ (user) + API socket. systemd socket activation (LISTEN_FDS) is supported; + without it the daemon binds the socket itself. + +# NOTES + +Materialization attempts a reflink (FICLONE) before falling back to a +byte copy; store directory and materialization targets must share a +filesystem (btrfs/XFS) for reflinks to work. Store files are immutable +and never hardlinked out. + +DSCP marking may require CAP_NET_ADMIN on some systems and is silently +skipped where unavailable. + +# SEE ALSO + +*varde-ctl*(1) diff --git a/dist/varde.service b/dist/varde.service new file mode 100644 index 0000000..19a9e8d --- /dev/null +++ b/dist/varde.service @@ -0,0 +1,38 @@ +# varde system service. Pairs with varde.socket for socket activation; +# also runs standalone (the daemon binds its own socket when not +# activated). +[Unit] +Description=varde content-addressed blob mirror daemon +Documentation=man:varde(8) +After=network.target +Requires=varde.socket + +[Service] +Type=notify +ExecStart=/usr/bin/varde-daemon --system +Restart=on-failure + +# Hardening +DynamicUser=yes +StateDirectory=varde +RuntimeDirectory=varde +NoNewPrivileges=yes +ProtectSystem=strict +ProtectHome=yes +PrivateTmp=yes +PrivateDevices=yes +ProtectKernelTunables=yes +ProtectKernelModules=yes +ProtectControlGroups=yes +RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 AF_NETLINK +RestrictNamespaces=yes +RestrictRealtime=yes +RestrictSUIDSGID=yes +LockPersonality=yes +MemoryDenyWriteExecute=yes +SystemCallArchitectures=native +SystemCallFilter=@system-service +CapabilityBoundingSet= + +[Install] +WantedBy=multi-user.target diff --git a/dist/varde.socket b/dist/varde.socket new file mode 100644 index 0000000..fdafb64 --- /dev/null +++ b/dist/varde.socket @@ -0,0 +1,10 @@ +# Socket activation for the varde system daemon. +[Unit] +Description=varde API socket + +[Socket] +ListenStream=/run/varde/varde.sock +SocketMode=0666 + +[Install] +WantedBy=sockets.target diff --git a/docs/milestone-5.md b/docs/milestone-5.md new file mode 100644 index 0000000..52677b1 --- /dev/null +++ b/docs/milestone-5.md @@ -0,0 +1,42 @@ +# Milestone 5 — Citizenship + +## What landed + +- **Metered awareness** (`metered` feature, default on): NetworkManager's + `Metered` property is polled over D-Bus every 30 s; while metered (or + guessed metered) the daemon serves nothing (connections closed at + accept) and defers every fetch. Pins stay recorded; mdns + re-announcements retrigger auto-sync once unmetered. Builds without + D-Bus via `--no-default-features` (always unmetered then). + `VARDE_FORCE_METERED=true` forces the state — a manual kill switch and + the test hook. +- **DSCP CS1** on the endpoint's UDP sockets, best effort. iroh does not + expose its socket fds, so we match the endpoint's bound ports against + `/proc/net/udp{,6}` inodes and walk `/proc/self/fd` — Linux-only like + the daemon itself. Failures are logged at debug and ignored, as the + spec allows. +- **systemd socket activation**: `LISTEN_FDS` (fd 3) is adopted when + present, otherwise the daemon binds its own socket — non-systemd + distros work identically. Readiness is signaled with a hand-rolled + `sd_notify(READY=1)` (abstract and path sockets), no libsystemd + dependency. Activation sockets are not unlinked at shutdown; systemd + owns them. +- **dist/**: hardened system units (`DynamicUser`, `ProtectSystem=strict`, + `StateDirectory`/`RuntimeDirectory`, `NoNewPrivileges`, restricted + address families and syscalls), user units, commented example + `config.toml`, scdoc man pages (`varde.8`, `varde-ctl.1`, validated + with scdoc), and an explicitly untested `PKGBUILD` skeleton for Arch. + +## Decisions + +- **Type=notify + Requires=varde.socket** so systemd tracks readiness + precisely and activation works out of the box; the same binary still + runs under `ExecStart` alone. +- **Metered gating sits at the two choke points** (connection accept, + fetch queue) rather than inside the rate limiter, so the posture is + binary and auditable: metered ⇒ zero bytes moved, period. +- **Polling NM instead of subscribing to PropertiesChanged**: 30 s of + staleness is acceptable for a background mirror and the code stays + ~40 lines with graceful degradation when NM or D-Bus is absent. +- `varde-ctl` already had `--json` and human output from milestone 1; + the polish here was man-page coverage of every command. diff --git a/varde-daemon/Cargo.toml b/varde-daemon/Cargo.toml index beed125..5be0667 100644 --- a/varde-daemon/Cargo.toml +++ b/varde-daemon/Cargo.toml @@ -22,6 +22,8 @@ n0-future = "0.1" async-channel = "2" futures-lite = "2" bytes = "1" +# Feature-gated D-Bus client for NetworkManager metered status. +zbus = { version = "5", optional = true, default-features = false, features = ["tokio"] } iroh-io = { workspace = true } reflink-copy = { workspace = true } anyhow = { workspace = true } @@ -33,6 +35,12 @@ toml = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +[features] +# Metered-connection detection via NetworkManager over D-Bus. On by +# default; disable for systems without D-Bus (they run as unmetered). +default = ["metered"] +metered = ["dep:zbus"] + [dev-dependencies] tempfile = "3" proptest = "1" diff --git a/varde-daemon/src/daemon.rs b/varde-daemon/src/daemon.rs index 6780f83..fb5e9b8 100644 --- a/varde-daemon/src/daemon.rs +++ b/varde-daemon/src/daemon.rs @@ -58,6 +58,7 @@ impl Daemon { let open_filter: crate::shaped::ServeFilter = Arc::new(move |hash| filter_set.read().expect("open set lock").contains(hash)); + let metered = crate::metered::start(); let transfer = Transfer::start( &config, &store, @@ -65,6 +66,7 @@ impl Daemon { open_filter, store.pool_handle(), events.clone(), + metered, ) .await?; diff --git a/varde-daemon/src/dscp.rs b/varde-daemon/src/dscp.rs new file mode 100644 index 0000000..69aabb4 --- /dev/null +++ b/varde-daemon/src/dscp.rs @@ -0,0 +1,102 @@ +//! Best-effort DSCP marking of the endpoint's UDP sockets. +//! +//! Background transfer traffic is marked CS1 (low priority) so routers +//! that honor DSCP deprioritize it. iroh does not expose its socket fds, +//! so this walks `/proc/self/fd` for the UDP sockets bound to the +//! endpoint's ports — Linux-only by construction, like the daemon. +//! Failures are logged at debug and ignored: some environments strip or +//! forbid DSCP (may require CAP_NET_ADMIN), and that's fine. + +use std::net::SocketAddr; +use std::os::fd::RawFd; + +use tracing::debug; + +/// DSCP CS1 (001000) in the TOS byte. +const TOS_CS1: u32 = 0x20; + +extern "C" { + fn setsockopt(fd: i32, level: i32, name: i32, value: *const u32, len: u32) -> i32; +} + +const IPPROTO_IP: i32 = 0; +const IP_TOS: i32 = 1; +const IPPROTO_IPV6: i32 = 41; +const IPV6_TCLASS: i32 = 67; + +/// Mark every UDP socket bound to one of `addrs` with DSCP CS1. +pub fn mark_endpoint_sockets(addrs: &[SocketAddr]) { + let ports: Vec = addrs.iter().map(|a| a.port()).collect(); + let inodes = udp_socket_inodes(&ports); + if inodes.is_empty() { + debug!("dscp: no matching udp sockets found"); + return; + } + let mut marked = 0; + for fd in socket_fds(&inodes) { + // Set both levels; one of them will apply depending on family. + let v4 = unsafe { setsockopt(fd, IPPROTO_IP, IP_TOS, &TOS_CS1, 4) }; + let v6 = unsafe { setsockopt(fd, IPPROTO_IPV6, IPV6_TCLASS, &TOS_CS1, 4) }; + if v4 == 0 || v6 == 0 { + marked += 1; + } + } + debug!(marked, "dscp: marked endpoint sockets CS1"); +} + +/// Inodes of UDP sockets locally bound to one of `ports`, from +/// /proc/net/udp{,6}. Format: whitespace columns, local_address is +/// `HEXIP:HEXPORT` in column 1, inode in column 9. +fn udp_socket_inodes(ports: &[u16]) -> Vec { + let mut inodes = Vec::new(); + for table in ["/proc/net/udp", "/proc/net/udp6"] { + let Ok(text) = std::fs::read_to_string(table) else { + continue; + }; + for line in text.lines().skip(1) { + let fields: Vec<&str> = line.split_whitespace().collect(); + let (Some(local), Some(inode)) = (fields.get(1), fields.get(9)) else { + continue; + }; + let Some((_, port_hex)) = local.rsplit_once(':') else { + continue; + }; + let Ok(port) = u16::from_str_radix(port_hex, 16) else { + continue; + }; + if ports.contains(&port) { + if let Ok(inode) = inode.parse() { + inodes.push(inode); + } + } + } + } + inodes +} + +/// Our process's fds that are sockets with one of the given inodes. +fn socket_fds(inodes: &[u64]) -> Vec { + let mut fds = Vec::new(); + let Ok(entries) = std::fs::read_dir("/proc/self/fd") else { + return fds; + }; + for entry in entries.flatten() { + let Ok(target) = std::fs::read_link(entry.path()) else { + continue; + }; + let target = target.to_string_lossy(); + let Some(inode) = target + .strip_prefix("socket:[") + .and_then(|s| s.strip_suffix(']')) + .and_then(|s| s.parse::().ok()) + else { + continue; + }; + if inodes.contains(&inode) { + if let Ok(fd) = entry.file_name().to_string_lossy().parse::() { + fds.push(fd); + } + } + } + fds +} diff --git a/varde-daemon/src/lib.rs b/varde-daemon/src/lib.rs index 5a80090..a8ab36b 100644 --- a/varde-daemon/src/lib.rs +++ b/varde-daemon/src/lib.rs @@ -5,7 +5,9 @@ pub mod config; pub mod daemon; +pub mod dscp; pub mod meta; +pub mod metered; pub mod server; pub mod shaped; pub mod store; diff --git a/varde-daemon/src/main.rs b/varde-daemon/src/main.rs index c0275e3..f1c03db 100644 --- a/varde-daemon/src/main.rs +++ b/varde-daemon/src/main.rs @@ -82,16 +82,25 @@ fn main() -> Result<()> { async fn run(config: Config) -> Result<()> { let socket_path = config.socket_path.clone(); let daemon = daemon::Daemon::open(config).await?; - let listener = server::bind_socket(&socket_path)?; + // Prefer a systemd activation socket; bind ourselves otherwise so + // non-systemd distros work identically. + let (listener, activated) = match server::activation_listener()? { + Some(listener) => (listener, true), + None => (server::bind_socket(&socket_path)?, false), + }; + server::notify_ready(); let result = tokio::select! { r = server::serve(listener, daemon.clone()) => r, r = shutdown_signal() => r.map(|signal| info!(signal, "shutting down")), }; - // Close the endpoint, flush the store, remove the socket. + // Close the endpoint and flush the store. The socket file is ours to + // remove only when we bound it (systemd owns activation sockets). daemon.shutdown().await; - let _ = std::fs::remove_file(&socket_path); + if !activated { + let _ = std::fs::remove_file(&socket_path); + } result } diff --git a/varde-daemon/src/metered.rs b/varde-daemon/src/metered.rs new file mode 100644 index 0000000..51ae86a --- /dev/null +++ b/varde-daemon/src/metered.rs @@ -0,0 +1,99 @@ +//! Metered-connection awareness. +//! +//! When the active connection is metered, varde stops moving bytes: +//! no background fetching, no serving. Detection queries +//! NetworkManager's `Metered` property over D-Bus and is feature-gated +//! (`metered`, on by default) so the daemon still builds and runs on +//! systems without D-Bus — those simply always count as unmetered. +//! +//! `VARDE_FORCE_METERED=true` forces the metered state regardless of +//! NetworkManager; useful for testing and as a manual kill switch. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; + +/// Shared metered state, cheap to read on every transfer decision. +#[derive(Debug, Clone, Default)] +pub struct MeteredState(Arc); + +impl MeteredState { + pub fn is_metered(&self) -> bool { + self.0.load(Ordering::Relaxed) + } + + fn set(&self, metered: bool) { + let was = self.0.swap(metered, Ordering::Relaxed); + if was != metered { + tracing::info!(metered, "metered state changed"); + } + } +} + +/// Start metered detection and return the shared state. +pub fn start() -> MeteredState { + let state = MeteredState::default(); + if std::env::var("VARDE_FORCE_METERED").as_deref() == Ok("true") { + tracing::warn!("VARDE_FORCE_METERED=true: all transfers disabled"); + state.set(true); + return state; + } + #[cfg(feature = "metered")] + { + let state = state.clone(); + tokio::spawn(async move { poll_network_manager(state).await }); + } + state +} + +/// NetworkManager's NM_METERED enum: 1 = yes, 3 = guess-yes. +#[cfg(feature = "metered")] +async fn poll_network_manager(state: MeteredState) { + use std::time::Duration; + + let connection = match zbus::Connection::system().await { + Ok(c) => c, + Err(e) => { + tracing::debug!(error = %e, "no D-Bus system bus; assuming unmetered"); + return; + } + }; + let proxy = match zbus::Proxy::new( + &connection, + "org.freedesktop.NetworkManager", + "/org/freedesktop/NetworkManager", + "org.freedesktop.NetworkManager", + ) + .await + { + Ok(p) => p, + Err(e) => { + tracing::debug!(error = %e, "NetworkManager proxy failed; assuming unmetered"); + return; + } + }; + + loop { + match proxy.get_property::("Metered").await { + Ok(value) => state.set(value == 1 || value == 3), + Err(e) => { + tracing::debug!(error = %e, "reading Metered property"); + } + } + tokio::time::sleep(Duration::from_secs(30)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_unmetered_and_flips() { + let state = MeteredState::default(); + assert!(!state.is_metered()); + state.set(true); + assert!(state.is_metered()); + state.set(false); + assert!(!state.is_metered()); + } +} diff --git a/varde-daemon/src/server.rs b/varde-daemon/src/server.rs index 441db22..47a8882 100644 --- a/varde-daemon/src/server.rs +++ b/varde-daemon/src/server.rs @@ -11,6 +11,57 @@ use varde_proto::{Envelope, ErrorCode, Request}; use crate::daemon::Daemon; +/// A listener passed in by systemd socket activation (`LISTEN_FDS`), +/// if any. Protocol: fds start at 3; `LISTEN_PID`, when set, must be us. +pub fn activation_listener() -> Result> { + let Ok(listen_fds) = std::env::var("LISTEN_FDS") else { + return Ok(None); + }; + if let Ok(pid) = std::env::var("LISTEN_PID") { + if pid != std::process::id().to_string() { + return Ok(None); + } + } + let n: u32 = listen_fds.parse().context("parsing LISTEN_FDS")?; + anyhow::ensure!(n == 1, "expected exactly one activation fd, got {n}"); + // SAFETY: fd 3 is the first activation fd per the LISTEN_FDS + // protocol; we take sole ownership of it. + let std_listener = unsafe { + use std::os::fd::FromRawFd; + std::os::unix::net::UnixListener::from_raw_fd(3) + }; + std_listener + .set_nonblocking(true) + .context("setting activation socket nonblocking")?; + let listener = UnixListener::from_std(std_listener).context("adopting activation socket")?; + info!("using systemd activation socket"); + Ok(Some(listener)) +} + +/// Tell the service manager we are ready (`sd_notify(READY=1)`), +/// hand-rolled to avoid a libsystemd dependency. No-op without +/// `NOTIFY_SOCKET`. +pub fn notify_ready() { + let Some(path) = std::env::var_os("NOTIFY_SOCKET") else { + return; + }; + let result = (|| -> std::io::Result<()> { + let socket = std::os::unix::net::UnixDatagram::unbound()?; + let bytes = path.as_encoded_bytes(); + if let Some(name) = bytes.strip_prefix(b"@") { + use std::os::linux::net::SocketAddrExt; + let addr = std::os::unix::net::SocketAddr::from_abstract_name(name)?; + socket.send_to_addr(b"READY=1", &addr)?; + } else { + socket.send_to(b"READY=1", &path)?; + } + Ok(()) + })(); + if let Err(e) = result { + debug!(error = %e, "sd_notify failed"); + } +} + /// Bind the API socket, replacing a stale socket file if the previous /// daemon did not shut down cleanly. pub fn bind_socket(path: &Path) -> Result { diff --git a/varde-daemon/src/transfer.rs b/varde-daemon/src/transfer.rs index 90608e3..3cd0507 100644 --- a/varde-daemon/src/transfer.rs +++ b/varde-daemon/src/transfer.rs @@ -27,6 +27,7 @@ use varde_proto::{Direction, Event}; use crate::config::Config; use crate::meta::Meta; +use crate::metered::MeteredState; use crate::shaped::{ServeFilter, ShapedStore, TokenBucket}; use crate::store::BlobStore; @@ -39,6 +40,7 @@ pub struct Transfer { router: Router, downloader: Downloader, events: broadcast::Sender, + metered: MeteredState, discovery_rx: std::sync::Mutex>>, } @@ -54,6 +56,7 @@ struct GatedProvider { events: EventSender, pool: LocalPoolHandle, serving_enabled: bool, + metered: MeteredState, } impl ProtocolHandler for GatedProvider { @@ -61,8 +64,8 @@ impl ProtocolHandler for GatedProvider { let this = self.clone(); Box::pin(async move { let remote = connection.remote_node_id()?; - if !this.serving_enabled { - debug!(%remote, "serving disabled, closing connection"); + if !this.serving_enabled || this.metered.is_metered() { + debug!(%remote, "serving disabled or metered, closing connection"); connection.close(0u32.into(), b"serving disabled"); return Ok(()); } @@ -116,6 +119,7 @@ impl Transfer { open_filter: ServeFilter, pool: LocalPoolHandle, events: broadcast::Sender, + metered: MeteredState, ) -> Result { let secret = load_or_create_secret(&config.store_dir.join("secret.key"))?; let mut builder = Endpoint::builder().secret_key(secret.clone()); @@ -146,6 +150,11 @@ impl Transfer { let endpoint = builder.bind().await.context("binding iroh endpoint")?; info!(node_id = %endpoint.node_id(), discovery = config.discovery, "endpoint up"); + // Best-effort low-priority marking of our UDP traffic. + let (v4, v6) = endpoint.bound_sockets(); + let bound: Vec = std::iter::once(v4).chain(v6).collect(); + crate::dscp::mark_endpoint_sockets(&bound); + let upload_bucket = TokenBucket::new(config.max_upload_bytes_per_sec); let download_bucket = TokenBucket::new(config.max_download_bytes_per_sec); let shaped = ShapedStore::new(store.inner().clone(), upload_bucket, download_bucket); @@ -158,6 +167,7 @@ impl Transfer { events: provider_events, pool: pool.clone(), serving_enabled: config.max_upload_bytes_per_sec > 0, + metered: metered.clone(), }; let downloader = Downloader::new(shaped, endpoint.clone(), pool); @@ -170,6 +180,7 @@ impl Transfer { router, downloader, events, + metered, discovery_rx: std::sync::Mutex::new(Some(discovery_rx)), }) } @@ -200,6 +211,11 @@ impl Transfer { /// 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) { + if self.metered.is_metered() { + // The pin stays recorded; auto-sync retries once unmetered. + info!(hash = %content.hash.to_hex(), "metered connection: fetch deferred"); + return; + } let downloader = self.downloader.clone(); let events = self.events.clone(); let endpoint = self.endpoint.clone(); diff --git a/varde-daemon/tests/citizenship.rs b/varde-daemon/tests/citizenship.rs new file mode 100644 index 0000000..5b5d0cd --- /dev/null +++ b/varde-daemon/tests/citizenship.rs @@ -0,0 +1,215 @@ +//! Milestone 5: systemd socket activation, readiness notification, and +//! metered behavior. + +mod support; + +use std::os::fd::AsRawFd; +use std::os::unix::process::CommandExt; +use std::path::Path; +use std::process::Command; +use std::time::{Duration, Instant}; + +use varde_proto::{PinPolicy, Request, ResponseData}; + +extern "C" { + fn dup2(oldfd: i32, newfd: i32) -> i32; +} + +/// Spawn the daemon with an activation socket on fd 3, the LISTEN_FDS +/// protocol systemd uses. +#[test] +fn systemd_socket_activation() { + let dir = tempfile::tempdir().unwrap(); + let socket_path = dir.path().join("activated.sock"); + let listener = std::os::unix::net::UnixListener::bind(&socket_path).unwrap(); + let fd = listener.as_raw_fd(); + + std::fs::write(dir.path().join("no-config.toml"), "").unwrap(); + let mut command = Command::new(env!("CARGO_BIN_EXE_varde-daemon")); + command + .arg("--user") + .arg("--store") + .arg(dir.path().join("store")) + // Deliberately point --socket elsewhere: the activation fd must + // win over self-binding. + .arg("--socket") + .arg(dir.path().join("ignored.sock")) + .env("VARDE_CONFIG", dir.path().join("no-config.toml")) + .env("VARDE_DISCOVERY", "false") + .env("LISTEN_FDS", "1") + .env_remove("LISTEN_PID"); + // SAFETY: dup2 is async-signal-safe; we place the listener on fd 3 + // in the child, which also clears CLOEXEC on the duplicate. + unsafe { + command.pre_exec(move || { + if dup2(fd, 3) < 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + let mut child = command.spawn().expect("spawning daemon"); + + // The daemon must answer on the activation socket. + let deadline = Instant::now() + Duration::from_secs(10); + let mut answered = false; + while Instant::now() < deadline { + if std::os::unix::net::UnixStream::connect(&socket_path).is_ok() { + let mut client = support::Client::connect(&socket_path); + let reply = client.request(&Request::Status { hash: None }); + if reply.ok { + answered = true; + break; + } + } + std::thread::sleep(Duration::from_millis(50)); + } + let _ = child.kill(); + let _ = child.wait(); + assert!(answered, "daemon did not serve on the activation socket"); + assert!( + !dir.path().join("ignored.sock").exists(), + "daemon bound its own socket despite activation" + ); +} + +/// With NOTIFY_SOCKET set, the daemon reports READY=1 once serving. +#[test] +fn sd_notify_ready_is_sent() { + let dir = tempfile::tempdir().unwrap(); + let notify_path = dir.path().join("notify.sock"); + let notify = std::os::unix::net::UnixDatagram::bind(¬ify_path).unwrap(); + notify + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + + std::fs::write(dir.path().join("no-config.toml"), "").unwrap(); + let mut child = Command::new(env!("CARGO_BIN_EXE_varde-daemon")) + .arg("--user") + .arg("--store") + .arg(dir.path().join("store")) + .arg("--socket") + .arg(dir.path().join("varde.sock")) + .env("VARDE_CONFIG", dir.path().join("no-config.toml")) + .env("VARDE_DISCOVERY", "false") + .env("NOTIFY_SOCKET", ¬ify_path) + .spawn() + .expect("spawning daemon"); + + let mut buf = [0u8; 64]; + let result = notify.recv(&mut buf); + let _ = child.kill(); + let _ = child.wait(); + let n = result.expect("no readiness notification received"); + let message = std::str::from_utf8(&buf[..n]).unwrap(); + assert!( + message.contains("READY=1"), + "unexpected notify payload: {message}" + ); +} + +fn wait_complete(socket: &Path, hash: &str, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let mut client = support::Client::connect(socket); + let reply = client.request(&Request::Status { + hash: Some(hash.to_string()), + }); + if let Some(ResponseData::HashStatus(s)) = reply.data { + if s.complete { + return true; + } + } + std::thread::sleep(Duration::from_millis(100)); + } + false +} + +/// A metered daemon serves nothing, even openly exported content. +#[test] +fn metered_daemon_stops_serving() { + let provider_dir = tempfile::tempdir().unwrap(); + let provider = + support::spawn_daemon_with_env(provider_dir.path(), &[("VARDE_FORCE_METERED", "true")]); + let fetcher = support::spawn_daemon(); + let mut provider_client = support::Client::connect(&provider.socket); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("f.bin"); + std::fs::write(&src, vec![4u8; 200_000]).unwrap(); + let reply = provider_client.request(&Request::Add { + path: src.display().to_string(), + recursive: false, + }); + let hash = match reply.data { + Some(ResponseData::Added { hash, .. }) => hash, + other => panic!("unexpected add reply: {other:?}"), + }; + let reply = provider_client.request(&Request::Pin { + hash: hash.clone(), + policy: PinPolicy::default(), + }); + assert!(reply.ok); + let reply = provider_client.request(&Request::TicketExport { hash: hash.clone() }); + let ticket = match reply.data { + Some(ResponseData::Ticket { ticket }) => ticket, + other => panic!("unexpected ticket reply: {other:?}"), + }; + + let reply = fetcher_client.request(&Request::TicketImport { + ticket, + pin_policy: PinPolicy::default(), + }); + assert!(reply.ok, "import should be accepted (pin recorded)"); + assert!( + !wait_complete(&fetcher.socket, &hash, Duration::from_secs(8)), + "metered provider must not serve" + ); +} + +/// A metered daemon defers its own fetches too. +#[test] +fn metered_daemon_defers_fetching() { + let provider = support::spawn_daemon(); + let fetcher_dir = tempfile::tempdir().unwrap(); + let fetcher = + support::spawn_daemon_with_env(fetcher_dir.path(), &[("VARDE_FORCE_METERED", "true")]); + let mut provider_client = support::Client::connect(&provider.socket); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("f.bin"); + std::fs::write(&src, vec![6u8; 200_000]).unwrap(); + let reply = provider_client.request(&Request::Add { + path: src.display().to_string(), + recursive: false, + }); + let hash = match reply.data { + Some(ResponseData::Added { hash, .. }) => hash, + other => panic!("unexpected add reply: {other:?}"), + }; + assert!( + provider_client + .request(&Request::Pin { + hash: hash.clone(), + policy: PinPolicy::default(), + }) + .ok + ); + let reply = provider_client.request(&Request::TicketExport { hash: hash.clone() }); + let ticket = match reply.data { + Some(ResponseData::Ticket { ticket }) => ticket, + other => panic!("unexpected ticket reply: {other:?}"), + }; + + let reply = fetcher_client.request(&Request::TicketImport { + ticket, + pin_policy: PinPolicy::default(), + }); + assert!(reply.ok, "import records the pin even while metered"); + assert!( + !wait_complete(&fetcher.socket, &hash, Duration::from_secs(8)), + "metered fetcher must not download" + ); +}