From b64b768c41f29b3e622f88c116c2ddecad535411 Mon Sep 17 00:00:00 2001 From: Louis Thibault Date: Sun, 23 Aug 2026 10:23:06 +0300 Subject: [PATCH] feat: add host-owned IPNS deployments --- .github/workflows/rust.yml | 2 + CHANGELOG.md | 14 + Cargo.lock | 94 ++- Cargo.toml | 3 + Makefile | 16 +- crates/ipfs/src/lib.rs | 53 -- crates/rpc/Cargo.toml | 3 + crates/rpc/src/keys.rs | 143 +++- diagrams/architecture-map.json | 18 +- doc/architecture.md | 53 +- doc/cli.md | 51 +- doc/deployment.md | 107 ++- doc/images.md | 21 +- doc/keys.md | 45 +- doc/routing.md | 24 + src/cli/daemon_cmd.rs | 12 +- src/cli/main.rs | 300 ++++++-- src/deployment/mod.rs | 118 +++ src/ipns.rs | 1279 +++++++++++++++++++++++++++++++ src/lib.rs | 2 + src/stem/ipns.rs | 646 ++++++++++++++++ src/stem/mod.rs | 1 + tests/cli_daemon_integration.rs | 28 +- tests/ipns_kubo_interop.rs | 268 +++++++ 24 files changed, 3118 insertions(+), 183 deletions(-) create mode 100644 src/ipns.rs create mode 100644 src/stem/ipns.rs create mode 100644 tests/ipns_kubo_interop.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 628e0cf6..8ae25a05 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -270,6 +270,8 @@ jobs: - name: Run host workspace tests (hot cache from build job) run: cargo test --workspace + env: + WW_TEST_REQUIRE_KUBO: "1" - name: Smoke-test Chess authority proof command run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index c6de80c2..3ea0abf2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] ### Changed +- **Wetware now owns default IPNS signing and can follow an IPNS deployment + Stem.** `ww run --ipns-stem ` uses locally verified raw records as the + authoritative deployment source. Signed EOL revokes the current generation + even during a routing outage, while durable raw-record watermarks reject + rollback across restarts. The installed daemon derives its default IPNS name + from `~/.ww/identity`, signs locally, persists before publication, and + republishes through Kubo HTTP Routing V1. Kubo 0.33 operators must enable + `Gateway.ExposeRoutingAPI` and use the Gateway listener, which defaults to + `http://localhost:8080` and can be set with `IPFS_ROUTING_API` or + `--ipns-routing-url`. Wetware no longer creates or requires Kubo's `"ww"` + signing key. Guest `Routing.publish` is unchanged. `rust-ipns` is temporarily + pinned to reviewed commit `02c5ae7bf3f9568c7dbbb1308ae9299cfc7ba2d9` + pending upstream PR #503 or a release that contains its V2-only validation + fix. - **Default daemon images no longer publish private host state.** Installed daemons import only `~/.ww/fhs` as their default deployment image. Identity and namespace configuration remain host-side. Updates rewrite vulnerable diff --git a/Cargo.lock b/Cargo.lock index 48189ca3..8d8b8b75 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1347,6 +1347,15 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" +[[package]] +name = "cbor4ii" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b544cf8c89359205f4f990d0e6f3828db42df85b5dac95d09157a250eb0749c4" +dependencies = [ + "serde", +] + [[package]] name = "cc" version = "1.2.62" @@ -1512,6 +1521,8 @@ checksum = "21a304f95f84d169a6f31c4d0a30d784643aaa0bbc9c1e449a2c23e963ec4971" dependencies = [ "multibase", "multihash", + "serde", + "serde_bytes", "unsigned-varint 0.8.0", ] @@ -2744,11 +2755,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 6.0.0", "rand_core 0.10.1", "wasip2", "wasip3", + "wasm-bindgen", ] [[package]] @@ -3356,6 +3369,17 @@ dependencies = [ "ww-cache", ] +[[package]] +name = "ipld-core" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "090f624976d72f0b0bb71b86d58dc16c15e069193067cb3a3a09d655246cbbda" +dependencies = [ + "cid", + "serde", + "serde_bytes", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -3701,9 +3725,9 @@ dependencies = [ [[package]] name = "libp2p-identity" -version = "0.2.13" +version = "0.2.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0c7892c221730ba55f7196e98b0b8ba5e04b4155651736036628e9f73ed6fc3" +checksum = "9525f3831544f7ae497bde79adf114ef127b0fbbb97edbbf692a80408636421c" dependencies = [ "asn1_der", "bs58", @@ -3711,7 +3735,7 @@ dependencies = [ "hkdf", "k256", "multihash", - "quick-protobuf", + "prost", "rand 0.8.6", "ring", "sha2", @@ -4213,6 +4237,7 @@ version = "0.19.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "577c63b00ad74d57e8c9aa870b5fccebf2fd64a308a5aee9f1bb88e4aea19447" dependencies = [ + "serde", "unsigned-varint 0.8.0", ] @@ -4854,6 +4879,29 @@ dependencies = [ "unarray", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools 0.10.5", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "pulley-interpreter" version = "45.0.0" @@ -5279,6 +5327,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "tempfile", "tokio", "tokio-util", "tracing", @@ -5337,6 +5386,22 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48fd7bd8a6377e15ad9d42a8ec25371b94ddc67abe7c8b9127bec79bebaaae18" +[[package]] +name = "rust-ipns" +version = "0.9.0" +source = "git+https://github.com/wetware/rust-ipfs?rev=02c5ae7bf3f9568c7dbbb1308ae9299cfc7ba2d9#02c5ae7bf3f9568c7dbbb1308ae9299cfc7ba2d9" +dependencies = [ + "bytes", + "chrono", + "getrandom 0.4.2", + "ipld-core", + "libp2p-identity", + "multihash", + "quick-protobuf", + "serde", + "serde_ipld_dagcbor", +] + [[package]] name = "rustc-demangle" version = "0.1.27" @@ -5574,6 +5639,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_bytes" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8" +dependencies = [ + "serde", + "serde_core", +] + [[package]] name = "serde_core" version = "1.0.228" @@ -5594,6 +5669,18 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "serde_ipld_dagcbor" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46182f4f08349a02b45c998ba3215d3f9de826246ba02bb9dddfe9a2a2100778" +dependencies = [ + "cbor4ii", + "ipld-core", + "scopeguard", + "serde", +] + [[package]] name = "serde_json" version = "1.0.150" @@ -8003,6 +8090,7 @@ dependencies = [ "rand 0.9.4", "reqwest", "rpc", + "rust-ipns", "serde", "serde_json", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index 8550109a..388e9f93 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -92,6 +92,9 @@ authority = { package = "wetware-authority", path = "crates/authority" } atom = { path = "crates/atom" } async-trait = { workspace = true } reqwest = { workspace = true, features = ["rustls-tls", "json"] } +# Temporary exact pin for V2-only record verification. Remove after +# https://github.com/dariusc93/rust-ipfs/pull/503 ships in a rust-ipns release. +rust-ipns = { git = "https://github.com/wetware/rust-ipfs", rev = "02c5ae7bf3f9568c7dbbb1308ae9299cfc7ba2d9", default-features = false, features = ["ed25519", "rsa"] } cache = { package = "ww-cache", path = "crates/cache" } cell = { path = "crates/cell" } rpc = { path = "crates/rpc" } diff --git a/Makefile b/Makefile index b4119a40..bef04faa 100644 --- a/Makefile +++ b/Makefile @@ -82,18 +82,12 @@ IPNS_KEY ?= # Best-effort publish: runs as part of `make all`. If Kubo isn't running, # the build continues without a CID (HostPathLoader fallback). -# Reads IPNS key from ~/.ww/etc/ns/ww if available (provisioned by `ww perform install`). +# Default IPNS publication is host-owned and runs through `ww perform update` +# plus the daemon republisher. This build target only imports the tree. try-publish-std: std - @KEY=$$(grep '^ipns=' ~/.ww/etc/ns/ww 2>/dev/null | cut -d= -f2 | tr -d ' '); \ - if [ -n "$$KEY" ]; then \ - $(MAKE) publish-std IPNS_KEY=ww 2>/dev/null \ - && echo " std namespace published to IPFS" \ - || echo " std namespace publish skipped (Kubo not running)"; \ - else \ - $(MAKE) publish-std 2>/dev/null \ - && echo " std namespace published to IPFS (no IPNS key)" \ - || echo " std namespace publish skipped (Kubo not running)"; \ - fi + @$(MAKE) publish-std 2>/dev/null \ + && echo " std namespace published to IPFS (host IPNS publish is daemon-owned)" \ + || echo " std namespace publish skipped (Kubo not running)" publish-std: std @echo "Assembling std namespace tree..." diff --git a/crates/ipfs/src/lib.rs b/crates/ipfs/src/lib.rs index 9bc8234c..d6cf9011 100644 --- a/crates/ipfs/src/lib.rs +++ b/crates/ipfs/src/lib.rs @@ -692,59 +692,6 @@ impl HttpClient { .map(|s| s.to_string()) .ok_or_else(|| anyhow::anyhow!("name publish response missing Name field")) } - - /// List IPNS key names on the local Kubo node. - pub async fn key_list(&self) -> anyhow::Result> { - let url = format!("{}/api/v0/key/list", self.base_url); - let response = self - .http_client - .post(&url) - .send() - .await - .context("IPFS key list request failed")?; - let status = response.status(); - let body: serde_json::Value = response - .json() - .await - .context("Failed to parse key list response")?; - if !status.is_success() { - let msg = body["Message"].as_str().unwrap_or("unknown error"); - anyhow::bail!("IPFS key list failed ({}): {}", status, msg); - } - let keys = body["Keys"] - .as_array() - .map(|arr| { - arr.iter() - .filter_map(|k| k["Name"].as_str().map(|s| s.to_string())) - .collect() - }) - .unwrap_or_default(); - Ok(keys) - } - - /// Generate a new Ed25519 IPNS key. Returns the key's peer ID. - pub async fn key_gen(&self, name: &str) -> anyhow::Result { - let url = format!("{}/api/v0/key/gen?arg={}&type=ed25519", self.base_url, name); - let response = self - .http_client - .post(&url) - .send() - .await - .context("IPFS key gen request failed")?; - let status = response.status(); - let body: serde_json::Value = response - .json() - .await - .context("Failed to parse key gen response")?; - if !status.is_success() { - let msg = body["Message"].as_str().unwrap_or("unknown error"); - anyhow::bail!("IPFS key gen failed ({}): {}", status, msg); - } - body["Id"] - .as_str() - .map(|s| s.to_string()) - .ok_or_else(|| anyhow::anyhow!("key gen response missing Id field")) - } } #[cfg(test)] diff --git a/crates/rpc/Cargo.toml b/crates/rpc/Cargo.toml index e0f0a41c..336388e5 100644 --- a/crates/rpc/Cargo.toml +++ b/crates/rpc/Cargo.toml @@ -30,3 +30,6 @@ tracing = { workspace = true } auth = { path = "../guest/auth" } ipfs = { path = "../ipfs" } authority = { package = "wetware-authority", path = "../authority" } + +[dev-dependencies] +tempfile = { workspace = true } diff --git a/crates/rpc/src/keys.rs b/crates/rpc/src/keys.rs index 0686a37f..fd009dbe 100644 --- a/crates/rpc/src/keys.rs +++ b/crates/rpc/src/keys.rs @@ -15,6 +15,8 @@ use anyhow::{bail, Context, Result}; use base58::{FromBase58, ToBase58}; use ed25519_dalek::SigningKey; use libp2p::identity::Keypair; +use std::io::Write; +use std::path::Path; /// Generate a new random Ed25519 signing key using the OS CSPRNG. pub fn generate() -> Result { @@ -64,13 +66,97 @@ pub fn load(path: &str) -> Result { /// Write a base58btc-encoded Ed25519 private key to disk. /// -/// Parent directories are created as needed. -pub fn save(sk: &SigningKey, path: &std::path::Path) -> Result<()> { - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .with_context(|| format!("create key directory: {}", parent.display()))?; +/// The replacement is written with mode 0600, synced, renamed atomically, and +/// followed by a parent-directory sync. Parent directories created by this +/// function use mode 0700 on Unix. +pub fn save(sk: &SigningKey, path: &Path) -> Result<()> { + atomic_write_private(path, encode(sk).as_bytes()) + .with_context(|| format!("write key: {}", path.display())) +} + +/// Atomically replace one trusted private-state file. +/// +/// Wetware supports one running process per private state directory. This +/// helper provides crash durability, not multiprocess coordination or +/// protection against malicious filesystem rollback. +pub fn atomic_write_private(path: &Path, bytes: &[u8]) -> Result<()> { + atomic_write_private_with(path, |file| file.write_all(bytes)) +} + +fn atomic_write_private_with( + path: &Path, + write: impl FnOnce(&mut std::fs::File) -> std::io::Result<()>, +) -> Result<()> { + use std::fs::OpenOptions; + + #[cfg(unix)] + use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; + + let parent = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let parent_existed = parent.exists(); + std::fs::create_dir_all(parent) + .with_context(|| format!("create private-state directory: {}", parent.display()))?; + #[cfg(unix)] + if !parent_existed { + std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700)) + .with_context(|| format!("restrict private-state directory: {}", parent.display()))?; + } + + let name = path + .file_name() + .and_then(|name| name.to_str()) + .context("private-state path has no UTF-8 file name")?; + let mut temporary = None; + for _ in 0..128 { + let candidate = parent.join(format!( + ".{name}.ww-tmp-{}-{:016x}", + std::process::id(), + rand::random::() + )); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o600); + match options.open(&candidate) { + Ok(file) => { + temporary = Some((candidate, file)); + break; + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(error) => { + return Err(error).with_context(|| { + format!("create temporary private-state file for {}", path.display()) + }); + } + } } - std::fs::write(path, encode(sk)).with_context(|| format!("write key: {}", path.display())) + let (temporary_path, mut file) = + temporary.context("could not allocate private-state temp file")?; + + let result = (|| -> Result<()> { + write(&mut file).with_context(|| format!("write {}", temporary_path.display()))?; + file.sync_all() + .with_context(|| format!("sync {}", temporary_path.display()))?; + drop(file); + std::fs::rename(&temporary_path, path).with_context(|| { + format!( + "replace private-state file {} with {}", + path.display(), + temporary_path.display() + ) + })?; + std::fs::File::open(parent) + .and_then(|directory| directory.sync_all()) + .with_context(|| format!("sync private-state directory: {}", parent.display()))?; + Ok(()) + })(); + if result.is_err() { + let _ = std::fs::remove_file(&temporary_path); + } + result } #[cfg(test)] @@ -116,4 +202,49 @@ mod tests { let short = [1u8; 16].to_base58(); assert!(decode(&short).is_err()); } + + #[cfg(unix)] + #[test] + fn save_is_restrictive_and_load_compatible() { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("private/identity"); + let key = generate().unwrap(); + save(&key, &path).unwrap(); + + assert_eq!( + std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + assert_eq!( + std::fs::metadata(path.parent().unwrap()) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o700 + ); + assert_eq!( + load(path.to_str().unwrap()).unwrap().to_bytes(), + key.to_bytes() + ); + } + + #[test] + fn interrupted_write_preserves_canonical_file() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("identity"); + std::fs::write(&path, b"canonical").unwrap(); + + let error = atomic_write_private_with(&path, |file| { + file.write_all(b"partial")?; + Err(std::io::Error::other("interrupted")) + }) + .unwrap_err(); + + assert!(format!("{error:#}").contains("interrupted")); + assert_eq!(std::fs::read(&path).unwrap(), b"canonical"); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 1); + } } diff --git a/diagrams/architecture-map.json b/diagrams/architecture-map.json index e8f24502..1c81b21c 100644 --- a/diagrams/architecture-map.json +++ b/diagrams/architecture-map.json @@ -13,9 +13,9 @@ "group": "THE HOST", "title": "The `ww` command", "subtitle": "Starts and controls a Wetware node.", - "files": ["src/cli/main.rs"], - "what": "The CLI parses `ww run`, `ww shell`, `ww perform`, namespace, and daemon commands. `ww run` selects configured root layers, starts host services, and applies process and operator policy to deployment outcomes.", - "built": "The binary lives at `src/cli/main.rs`. It combines `clap` command parsing with image loaders and the host runtime. Deployment owns generation transitions; the CLI retains terminal behavior and process exit policy." + "files": ["src/cli/main.rs", "src/cli/daemon_cmd.rs"], + "what": "The CLI parses `ww run`, `ww shell`, `ww perform`, namespace, and daemon commands. `ww run` selects configured root layers or one authoritative Atom/IPNS Stem, starts host services, and applies process and operator policy to deployment outcomes.", + "built": "The binary lives at `src/cli/main.rs`. It combines `clap` command parsing with image loaders and the host runtime. `src/cli/daemon_cmd.rs` preserves host-only identity, namespace, and Routing V1 configuration in generated services. Deployment owns generation transitions; the CLI retains terminal behavior and process exit policy." }, { "key": "loader", @@ -45,9 +45,9 @@ "group": "THE HOST", "title": "Host supervisor", "subtitle": "Owns the node service threads.", - "files": ["src/services.rs", "src/host.rs"], - "what": "The `Host` supervisor starts and stops the swarm, WAGI service, compiler, admin service, and executor pool. Long-running services own dedicated threads; deployment coordination runs in the main host runtime.", - "built": "`src/services.rs` defines `Service` and `Host`. `src/host.rs` owns libp2p behavior and swarm events. The executor pool uses local runtimes because Wasmtime stores are not `Send`." + "files": ["src/services.rs", "src/host.rs", "src/ipns.rs"], + "what": "The `Host` supervisor starts and stops the swarm, WAGI service, compiler, admin service, executor pool, and host-owned IPNS republisher. Long-running services own dedicated threads; deployment coordination runs in the main host runtime.", + "built": "`src/services.rs` defines `Service` and `Host`. `src/host.rs` owns libp2p behavior and swarm events. `src/ipns.rs` validates, persists, signs, transports, and republishes raw IPNS records. The executor pool uses local runtimes because Wasmtime stores are not `Send`." }, { "key": "epoch", @@ -61,9 +61,9 @@ "group": "THE HOST", "title": "Deployment lifecycle", "subtitle": "Turns optional Stem state into one rooted kernel generation.", - "files": ["src/stem/mod.rs", "src/stem/atom.rs", "src/deployment/mod.rs", "crates/atom/src/indexer.rs", "crates/authority/src/epoch.rs"], - "what": "Deployment may prepare one advisory Atom event candidate without changing authority. Finalized Source updates allocate epochs, revoke old authority, activate `CidTree`, and replace pid0.", - "built": "`src/stem` defines the backend-neutral Source contract and Atom finalized-depth adapter. The Atom indexer supplies lossy advisory candidates. `src/deployment` owns preparation, retry, supersession, pin ownership, activation, and generation coordination. `crates/authority` enforces the resulting local epoch." + "files": ["src/stem/mod.rs", "src/stem/atom.rs", "src/stem/ipns.rs", "src/deployment/mod.rs", "crates/atom/src/indexer.rs", "crates/authority/src/epoch.rs"], + "what": "Deployment may prepare one advisory Atom event candidate without changing authority. Finalized Atom or locally verified IPNS Source updates allocate epochs, revoke old authority, activate `CidTree`, and replace pid0. Signed IPNS expiry produces the same authoritative invalid-head transition.", + "built": "`src/stem` defines the backend-neutral Source contract, the Atom finalized-depth adapter, and the IPNS signed-record adapter with a durable ordering floor and EOL deadline. The Atom indexer supplies lossy advisory candidates. `src/deployment` owns preparation, retry, supersession, pin ownership, activation, and generation coordination. `crates/authority` enforces the resulting local epoch." }, { "key": "kernel", diff --git a/doc/architecture.md b/doc/architecture.md index 68d00fc5..6155ff5f 100644 --- a/doc/architecture.md +++ b/doc/architecture.md @@ -112,8 +112,13 @@ provenance do not become `Epoch.seq` and do not survive a process restart. An optional `stem::Source` establishes the authoritative mutable base head. The backend adapter applies its consistency rule before returning an update. The Atom adapter polls `Atom.head()` at `tip - confirmation_depth`; contract -events are not part of its correctness path. Without a Stem, deployment -composes the configured frozen layers at epoch `0` and starts no source task. +events are not part of its correctness path. The IPNS adapter fetches raw +signed records through Kubo HTTP Routing V1, validates them locally, and +maps an exact `/ipfs/` binding with no subpath to `Head`. Another validly +signed value maps to `InvalidHead`. Atom revision and IPNS sequence remain +private Source state. Neither value becomes `Epoch.seq`. +Without a Stem, deployment composes the configured frozen layers at epoch `0` +and starts no source task. `deployment` owns every transition. After it accepts a Source update, it first publishes the incremented epoch with `root: None`. That publication closes @@ -135,6 +140,21 @@ An `InvalidHead` is an authoritative transition. Deployment publishes waits for a later valid update. A Source error does not advance the epoch and does not revoke the current deployment. +The IPNS Source treats transport failure and signed expiry differently. A +fetch failure preserves current authority before the accepted record's EOL. +`Source::next()` races retrieval and retry against that EOL. If no valid +replacement arrives first, the Source emits one `InvalidHead`. Deployment then +applies the same revoke-first transition used for every authoritative invalid +head. A later valid record can establish a new rooted generation. + +The Source persists each accepted raw signed record before it returns an +update. The persisted record is an ordering floor across restarts. An expired +persisted record still rejects older records, but its value does not seed the +boot deployment. Exact duplicates, higher-order records with the same value, +and valid same-value EOL refreshes do not advance the deployment epoch. +Same-sequence records with different values are treated as publisher +equivocation and do not replace the accepted binding. + Deployment retries classified transient preparation failures with jittered exponential backoff. Readiness stays closed and the old PID0 stays terminated during retry. A newer authoritative update supersedes in-progress preparation @@ -167,6 +187,35 @@ The import is installed only on PID0's linker, is not a Cap'n Proto capability value, and therefore cannot be delegated to children or transferred over the network. +## Host-owned IPNS publication + +The default `ww` namespace uses `~/.ww/identity` as both the host identity and +the IPNS signing identity. The host Peer ID and canonical Base36 IPNS name +derive from the same public key. Rotating the identity changes both names. + +Wetware signs and verifies IPNS records through the pinned `rust-ipns` +implementation. Kubo transports the raw protobuf through +`GET` and `PUT /routing/v1/ipns/{name}`. Kubo does not receive the private key +and is not the cryptographic authority. The temporary `rust-ipns` git pin is +commit `02c5ae7bf3f9568c7dbbb1308ae9299cfc7ba2d9`, pending upstream PR #503 or a +release that contains the reviewed V2-only validation fix. + +Publisher and follower state lives below private `~/.ww/ipns/`, outside the +publishable `~/.ww/fhs/` tree. The canonical state object is the raw signed +record. Wetware assumes one running process per private state directory and +trusts local state against rollback. The storage layer does not coordinate +multiple writers or defend against malicious filesystem rollback. + +The host publisher reconciles its persisted record with valid network state, +persists a new signed record before `PUT`, and retries failed publication with +the same bytes. Changed values increment sequence. Same-value refresh keeps +sequence and extends EOL. A newer network record with a different value +advances the durable floor but stops publication with a single-writer conflict; +Wetware does not overwrite that record automatically. The host republisher uses +a 48-hour lifetime, five-minute TTL, four-hour refresh interval, one-minute +initial delay, and five-minute retry delay. A followed third-party IPNS Stem +never creates a publisher. + ## Fixed execution substrate Every child has local computation, args and environment selected at spawn, diff --git a/doc/cli.md b/doc/cli.md index af029fcd..41bf6ab7 100644 --- a/doc/cli.md +++ b/doc/cli.md @@ -50,6 +50,8 @@ Layers stack with per-file union; later layers win. | `--runtime-cache-policy` | `shared` | `shared`: same WASM bytes share Executor. `isolated`: always fresh. | | `--ipfs-url ` | `http://localhost:5001` | IPFS HTTP API endpoint. Also reads `IPFS_API` env. | | `--stem ` | none | Atom contract address (hex, 0x-prefixed). Enables authoritative deployment following. | +| `--ipns-stem ` | none | Follow an IPNS name as the authoritative deployment Stem. Conflicts with `--stem`. | +| `--ipns-routing-url ` | `http://localhost:8080` | Kubo Gateway listener for raw HTTP Routing V1 records. Also reads `IPFS_ROUTING_API`. | | `--rpc-url ` | `http://127.0.0.1:8545` | HTTP JSON-RPC for finalized-depth `eth_blockNumber` and `eth_call` polling. | | `--ws-url ` | `ws://127.0.0.1:8545` | Deprecated compatibility option. Atom following does not require event subscriptions. | | `--confirmation-depth ` | `6` | Chain depth used when reading authoritative `Atom.head()` state. | @@ -75,8 +77,27 @@ ww run /ipfs/QmHash... # Atom-backed deployment lifecycle ww run . --stem 0x1234...abcd --rpc-url http://rpc.example.com:8545 + +# IPNS-backed deployment lifecycle +ww run --ipns-stem k51qzi5uqu5... --ipns-routing-url http://localhost:8080 +``` + +The IPNS Stem accepts canonical Base36 names, legacy base58 Peer IDs, and +`/ipns/`-prefixed forms. It is not an `/ipns/...` image mount. The Stem follows +signed record changes and revokes the deployment at signed EOL when no valid +replacement exists. + +Kubo 0.33 does not expose Routing V1 on the Gateway by default. Configure and +restart Kubo before using an IPNS Stem or the installed default publisher: + +```sh +ipfs config --json Gateway.ExposeRoutingAPI true ``` +Use the Gateway listener, normally port 8080. Port 5001 is the administrative +RPC listener and does not serve this path. Wetware does not fall back to Kubo's +`name/resolve` or `name/publish` APIs for host-owned IPNS records. + ### Environment variables | Variable | Set by | Description | @@ -85,6 +106,7 @@ ww run . --stem 0x1234...abcd --rpc-url http://rpc.example.com:8545 | `WW_TTY` | host | Set to `1` when stdin is a terminal (triggers interactive shell mode) | | `WW_CELL_MODE` | host | Cell transport mode: `vat`, `raw`, `http`, or absent (kernel) | | `IPFS_API` | user | Default IPFS HTTP API endpoint | +| `IPFS_ROUTING_API` | user | Kubo Gateway listener for HTTP Routing V1. Default: `http://localhost:8080`. | | `WW_HTTP_ADMIN` | user | Admin listener address, or `off` to disable it | | `WW_CWASM_DIR` | operator | Optional directory for Wasmtime's native compilation cache. Wasmtime owns artifact compatibility; an unavailable directory falls back to uncached compilation. | | `WW_CWASM_CACHE_MAX_BYTES` | operator | Wasmtime cache cleanup threshold in bytes (default: `335544320`). This is a soft limit, so leave filesystem headroom. | @@ -169,8 +191,27 @@ Bootstrap `~/.ww` and the daemon. Idempotent: re-running skips completed steps, retries failed ones. 1. Creates `~/.ww` directory structure -2. Generates Ed25519 identity (if missing) -3. Registers background daemon (launchd/systemd) +2. Generates `~/.ww/identity` if missing +3. Derives the host Peer ID and default Base36 IPNS name from that identity +4. Writes the default namespace bootstrap and derived IPNS name +5. Registers the background daemon (launchd/systemd) + +The command does not create a Kubo `"ww"` signing key. Kubo never receives the +Wetware identity. + +### ww perform update + +Refresh embedded images, the default namespace, and the daemon definition. + +```sh +ww perform update +``` + +The command keeps `~/.ww/identity`, rewrites the default namespace with the +IPNS name derived from that identity, and passes the Routing V1 URL to the +daemon. If Kubo is running without Routing V1, the command returns an error +that includes the required Kubo configuration. The daemon signs and publishes +the namespace record when it starts. ### ww perform upgrade @@ -198,9 +239,13 @@ Register wetware as a user-level background service (launchd on macOS, systemd on Linux). ``` -ww daemon install [--identity PATH] [--listen MULTIADDR ...] [--images PATH...] +ww daemon install [--identity PATH] [--listen MULTIADDR ...] [--images PATH...] [--ipns-routing-url URL] ``` +`--ipns-routing-url` defaults to `http://localhost:8080` and reads +`IPFS_ROUTING_API`. Generated launchd and systemd definitions preserve the +selected value. + ### ww daemon uninstall Remove the platform service file. diff --git a/doc/deployment.md b/doc/deployment.md index 308b80b0..eba5a495 100644 --- a/doc/deployment.md +++ b/doc/deployment.md @@ -10,10 +10,11 @@ The host-side `deployment` module owns runtime deployment transitions. A deployment combines an optional authoritative Stem head with configured frozen root layers. Later layers override earlier layers during DAG composition. -The boot deployment always uses local epoch `0`. An Atom contract revision is -diagnostic Source state; it never becomes `Epoch.seq`. A host restart allocates -epoch `0` again. Static startup prepares the frozen layers through the same -path, publishes a rooted epoch `0`, and starts no Source task. +The boot deployment always uses local epoch `0`. Atom contract revisions and +IPNS record sequences are Source state; neither value becomes `Epoch.seq`. A +host restart allocates epoch `0` again. Static startup prepares the frozen +layers through the same path, publishes a rooted epoch `0`, and starts no +Source task. For a replacement, deployment publishes the new local epoch with `root: None` before root preparation. This publication revokes old authority. Deployment @@ -33,6 +34,62 @@ ownership. It does not unpin a CID that another live category references. `CidTree::swap_root` only changes the root and clears its in-memory directory cache. Deployment removes stale readdir stubs after rooted publication. +## IPNS Stem + +`ww run --ipns-stem ` selects an IPNS name as the authoritative mutable +base. `--ipns-stem` and the Atom `--stem` option are mutually exclusive. The +name can be a canonical Base36 CIDv1 IPNS name, a legacy base58 Peer ID, or the +same name prefixed with `/ipns/`. + +The IPNS Source uses Kubo's Gateway listener for raw HTTP Routing V1 records: + +```text +GET /routing/v1/ipns/{canonical-base36-name} +Accept: application/vnd.ipfs.ipns-record +``` + +Wetware limits records to 10,240 bytes. `rust-ipns` decodes the protobuf, +verifies the signature and signer, and reads authenticated value, sequence, +EOL, and TTL semantics. The value must be exactly one `/ipfs/` binding. +Kubo supplies transport and routing only. + +The Source stores the accepted raw record at +`~/.ww/ipns/follow/.record`. Persistence completes +before the Source returns an update. On Unix, directories use mode `0700` and +record files use mode `0600`. The write path syncs a temporary file, renames +it, and syncs the parent directory. + +The persisted record is both the restart watermark and the ordering floor. A +valid unexpired record seeds boot. An expired record retains its ordering floor +but does not seed a deployment. Lower-order and exact-duplicate observations +are ignored. A higher-order record with the same value advances the watermark +without advancing the deployment epoch. A valid same-value refresh also +updates the stored record and EOL without a deployment transition. A +higher-order record with a different deployment CID emits `Head` after +persistence. A same-sequence record with a different value is publisher +equivocation; Wetware retains the accepted binding while valid. + +A validly signed higher-order record whose value is not exactly one +`/ipfs/` binding is also persisted before it emits `InvalidHead`. This is +an authoritative publisher selection, not network garbage. + +Malformed protobufs, invalid signatures, wrong signers, expired network +records, and transport failures do not revoke authority. Once an accepted +record reaches its signed EOL, the Source emits one `InvalidHead` even when +retrieval is retrying. Deployment advances its local epoch, publishes +`root: None`, and terminates the old PID0. The host remains alive. A later +valid record can recover through the ordinary deployment path. + +A Routing V1 `404` means that no record exists. Unsupported routes fail with +the Gateway configuration guidance above. Transport and server failures retry. +Unexpected media types, oversized bodies, cryptographic failures, and rejected +`PUT` responses remain distinct errors. Wetware does not select them as +authority. + +Wetware assumes one running process per `~/.ww` state directory. The local +state is trusted against rollback. The implementation does not provide +multiprocess coordination or secure monotonic storage. + ## Artifact publication ```text @@ -64,6 +121,48 @@ IPFS remains a required publication path while `scripts/install.sh` installs from IPNS. It must not be made non-blocking until a GitHub Release has been published and the installer has migrated to that distribution path. +## Default host IPNS publication + +`ww perform install` creates `~/.ww/identity`. `ww perform update` derives the +default IPNS name from that key, writes the name and immutable bootstrap CID to +`~/.ww/etc/ns/ww`, and renders the daemon with `--ipns-routing-url`. The +running daemon signs the configured bootstrap CID before namespace resolution. +If initial publication has a temporary failure, the daemon uses the immutable +bootstrap for that boot and retries publication in the background. + +The publisher stores its canonical raw record at +`~/.ww/ipns/publish/.record`. It reconciles valid newer +network state before creating a changed binding. It persists locally before +Routing V1 `PUT`, so a failed `PUT` can retry the same signed bytes. A changed +CID increments sequence. A same-CID refresh preserves sequence and extends +the signed EOL. If a newer network record selects a different value, Wetware +adopts its ordering floor but does not publish over it. Publication stops with +a single-writer conflict diagnostic for operator review. + +The republisher waits one minute after an initial success, then refreshes every +four hours. Failed publication retries after five minutes. Records use a +48-hour lifetime and five-minute TTL. Only the default name derived from the +host identity receives this lifecycle. Following a third-party IPNS Stem does +not publish that name. + +The default publisher no longer creates or uses Kubo's `"ww"` signing key. +An old `"ww"` key can remain in the Kubo keystore, but Wetware ignores it. +Kubo never needs `~/.ww/identity`. Guest `Routing.publish` remains a separate +Kubo-key-backed capability and is unchanged. + +Kubo 0.33 requires this one-time operator configuration: + +```sh +ipfs config --json Gateway.ExposeRoutingAPI true +# Restart Kubo after changing its configuration. +``` + +Wetware uses `http://localhost:8080` by default. Set `IPFS_ROUTING_API` or +`--ipns-routing-url` when the Gateway listener uses another address. Do not +point this option at the administrative RPC listener on port 5001. Wetware +fails with an actionable error when the listener does not expose Routing V1; +it does not fall back to Kubo `name/resolve`, `name/publish`, or key management. + ## Manual image promotion Production promotion is intentionally declarative: diff --git a/doc/images.md b/doc/images.md index 08dc3839..628e6bc2 100644 --- a/doc/images.md +++ b/doc/images.md @@ -30,9 +30,13 @@ as layers (later mounts override earlier ones): |------|---------| | Local path | `std/status` | | IPFS path | `/ipfs/QmAbc123...` | +| IPNS path | `/ipns/k51qzi5uqu5...` | | Layered | `ww run /ipfs/QmBase my-overlay` | Targeted mounts (`source:/guest/path`) are not accepted by backend virtual mode. +An `/ipns/...` mount is an image layer that Kubo resolves during composition. +It is distinct from `--ipns-stem`, which follows signed changes and replaces +the active deployment generation. ## Installed host layout @@ -45,6 +49,9 @@ the local image root: etc/ns/ # host-only namespace configuration logs/ # host logs run/ # host runtime state + ipns/ + follow/*.record # durable signed follower ordering floors + publish/*.record # durable signed records for the default publisher fhs/ # publishable local image root bin/status.wasm # status component used by the shipped Rust PID0 ``` @@ -57,7 +64,7 @@ Put intentional local image content under `~/.ww/fhs`, or pass an explicit image path to `ww daemon install`. Do not put private host state in an image root because Kubo can store, pin, and provide every imported file. -## On-chain coordination +## Stem coordination The `--stem` flag connects to an Atom contract on an EVM chain. The contract holds a monotonic head pointer (an IPFS CID). When the @@ -77,3 +84,15 @@ private to the Source and does not become `Epoch.seq`. This provides a coordination primitive across trust boundaries: multiple independent nodes watching the same contract will synchronize their agent lifecycle to the same on-chain state. + +`--ipns-stem ` selects an IPNS record instead of an Atom contract. The +two flags are mutually exclusive. Wetware fetches the raw signed record through +Kubo HTTP Routing V1 and validates it locally. A valid new `/ipfs/` value +uses the same revoke, prepare, swap, and replacement path described above. + +The signed record's EOL is authoritative. Transport failures before EOL keep +the current image active. EOL without a valid replacement emits +`InvalidHead`, removes the rooted epoch, and terminates PID0. An expired +persisted value does not seed boot, although its signed ordering floor remains +durable. See [deployment.md](deployment.md#ipns-stem) for ordering and recovery +semantics. diff --git a/doc/keys.md b/doc/keys.md index 08cedb02..9e2f9c61 100644 --- a/doc/keys.md +++ b/doc/keys.md @@ -9,9 +9,8 @@ concerns that were previously conflated: 1. **Node identity (Ed25519)** — the `PeerId` derived from the Ed25519 public key identifies the node on the p2p network and is used for Terminal - challenge-response signing. Ed25519 is libp2p's default and best-supported - key type: simpler (32-byte seed, 32-byte pubkey, no compressed/uncompressed - distinction), and the ecosystem default. + challenge-response and default IPNS publication. The canonical default IPNS + name is the Base36 CIDv1 form of the same `PeerId`. 2. **Operator identity (secp256k1)** — the Stem contract owner key. This is the one key that calls `setHead()` to advance the on-chain epoch. It is @@ -34,8 +33,8 @@ challenge-response auth uses `try_into_ed25519()`. Guest auth ### Key storage Keys are stored as **base58btc** (Bitcoin alphabet, ~44 characters for 32 bytes) -in a plain text file. Hex-encoded keys are also accepted on load for backward -compatibility. The default location is `~/.ww/identity`. +in a plain text file. Hex-encoded keys are not accepted. The default location +is `~/.ww/identity`. Rationale: - Denser than hex (44 vs 64 chars), no ambiguous characters (no 0/O/I/l). @@ -47,6 +46,34 @@ Rationale: image root is `~/.ww/fhs`, so the identity file does not enter the imported image DAG. An explicit image path can still expose any key inside that path. +The save path creates new private directories with mode `0700` on Unix. It +writes a mode-`0600` temporary file, syncs the file, renames it atomically, and +syncs the parent directory. The same helper stores durable raw IPNS records. +An interrupted write therefore does not truncate the canonical identity file. + +### Default IPNS identity + +`~/.ww/identity` intentionally serves two host-owned roles: + +```text +Ed25519 private key + ├─ libp2p host Peer ID + └─ default IPNS signer and name +``` + +Identity rotation changes both the Peer ID and the default IPNS name. +`ww perform update` rewrites `~/.ww/etc/ns/ww` with the derived canonical name. +The running daemon signs records locally and sends only raw signed records to +Kubo HTTP Routing V1. Wetware no longer creates or uses a Kubo `"ww"` signing +key for default publication. + +Signed publisher state lives at +`~/.ww/ipns/publish/.record`. Signed follower watermarks +live under `~/.ww/ipns/follow/`. Both directories remain outside +`~/.ww/fhs/`. Wetware assumes one running process per `~/.ww` state directory. +The local state is trusted against rollback; malicious filesystem rollback and +multiprocess coordination are out of scope. + ### Identity resolution `ww run` resolves the node identity in this order (first match wins): @@ -59,9 +86,9 @@ Rationale: Each time the host resolves the identity it logs the source at `INFO` level so the active source is always visible in the log output. -The ephemeral fallback is fine for local development and testing but means -the node's Peer ID and EVM address change on every restart. Use a persistent -key for any deployment that other nodes need to remember across restarts. +The ephemeral fallback is for local development and testing. It changes the +node Peer ID and derived IPNS name on every restart. Use a persistent key for +any deployment that other nodes need to remember across restarts. For daemon/service mode, pass identity through the same host flags/env: `--identity PATH` or `WW_IDENTITY=PATH`. @@ -83,7 +110,7 @@ ww run --identity ~/.ww/identity images/my-app ## File format ``` -# ~/.ww/identity — base58btc, ~44 chars (hex also accepted on load) +# ~/.ww/identity — base58btc, ~44 chars 6MRyAjQq8ud7hVNYcfnVPJqcVpscN5So8BhtHuGYqET5 ``` diff --git a/doc/routing.md b/doc/routing.md index 71fb06b9..dfde4de8 100644 --- a/doc/routing.md +++ b/doc/routing.md @@ -73,6 +73,30 @@ For IPNS updates, `publish` supports compare-and-set semantics: if `expected-current` is provided and does not match the currently resolved head, the call fails instead of silently overwriting. +## Host IPNS records and guest `Routing.publish` + +The default host publisher and an IPNS Stem do not use the guest `Routing` +capability. They use `src/ipns.rs` to sign or validate raw records locally and +use Kubo only for HTTP Routing V1 transport: + +```text +GET /routing/v1/ipns/{canonical-base36-name} +Accept: application/vnd.ipfs.ipns-record + +PUT /routing/v1/ipns/{canonical-base36-name} +Content-Type: application/vnd.ipfs.ipns-record +``` + +The default signer is `~/.ww/identity`. Kubo does not receive that private key. +Follower and publisher records remain private under `~/.ww/ipns/` and outside +the guest image root. + +Guest-visible `Routing.publish` is unchanged in this phase. It still calls +Kubo `name/resolve` for its optional compare-and-set check and Kubo +`name/publish` with the supplied key name. It does not use the host-owned +publisher state or grant access to `~/.ww/identity`. Redesign of that capability +remains deferred to ARCH-13. + ## Limitations - **Content routing only.** No key-value store (`putValue`/`getValue`) — deferred. diff --git a/src/cli/daemon_cmd.rs b/src/cli/daemon_cmd.rs index babce4f3..defb7356 100644 --- a/src/cli/daemon_cmd.rs +++ b/src/cli/daemon_cmd.rs @@ -22,6 +22,8 @@ pub(super) struct DaemonServiceConfig { pub images: Vec, /// Address (`host:port`) for the WAGI HTTP server. `None` disables WAGI. pub http_listen: Option, + /// Kubo Gateway listener that exposes HTTP Routing V1. + pub ipns_routing_url: String, } /// Register wetware as a user-level background service. @@ -33,6 +35,7 @@ pub(super) async fn daemon_install( identity: Option, listen: Vec, images: Vec, + ipns_routing_url: String, quiet: bool, ) -> Result { let home = dirs::home_dir().context("cannot determine home directory")?; @@ -76,6 +79,7 @@ pub(super) async fn daemon_install( namespace_roots, images: image_layers, http_listen: Some("127.0.0.1:2080".to_string()), + ipns_routing_url, }; // 3. Write platform service file. @@ -219,6 +223,11 @@ pub(super) fn write_launchd_plist( // Identity as a --identity flag (host-side only, not a guest mount). args.push(" --identity".to_string()); args.push(format!(" {identity_path}")); + args.push(" --ipns-routing-url".to_string()); + args.push(format!( + " {}", + config.ipns_routing_url + )); // WAGI HTTP listen addr (engagement starter kit: status endpoint on :2080). if let Some(ref addr) = config.http_listen { args.push(" --http-listen".to_string()); @@ -315,10 +324,11 @@ pub(super) fn write_systemd_unit( .collect::>() .join(" "); let exec_start = format!( - "{} run {} --identity {}{} {} {}", + "{} run {} --identity {} --ipns-routing-url {}{} {} {}", ww_bin.display(), listen_args, identity_path, + config.ipns_routing_url, http_listen_arg, namespace_args, positional.join(" "), diff --git a/src/cli/main.rs b/src/cli/main.rs index 3d7acb6a..875637ac 100644 --- a/src/cli/main.rs +++ b/src/cli/main.rs @@ -157,9 +157,22 @@ enum Commands { /// Atom contract address (hex, 0x-prefixed). Enables authoritative /// deployment following and authority revocation on head changes. - #[arg(long)] + #[arg(long, conflicts_with = "ipns_stem")] stem: Option, + /// IPNS name to follow as the authoritative deployment Stem. + #[arg(long, value_name = "IPNS_NAME", conflicts_with = "stem")] + ipns_stem: Option, + + /// HTTP Routing V1 Gateway listener used for raw IPNS records. + /// Kubo 0.33 requires Gateway.ExposeRoutingAPI=true. + #[arg( + long, + default_value = "http://localhost:8080", + env = "IPFS_ROUTING_API" + )] + ipns_routing_url: String, + /// HTTP JSON-RPC URL for finalized-depth Atom polling. #[arg(long, default_value = "http://127.0.0.1:8545")] rpc_url: String, @@ -367,6 +380,14 @@ enum DaemonAction { /// Image layers to run (local paths or IPFS CIDs). #[arg(long, value_name = "PATH")] images: Vec, + + /// HTTP Routing V1 Gateway listener for host-owned IPNS publication. + #[arg( + long, + default_value = "http://localhost:8080", + env = "IPFS_ROUTING_API" + )] + ipns_routing_url: String, }, /// Remove the platform service file. @@ -734,6 +755,8 @@ impl Commands { identity, insecure_ephemeral, stem, + ipns_stem, + ipns_routing_url, rpc_url, ws_url, confirmation_depth, @@ -777,6 +800,8 @@ impl Commands { wasm_debug, kernel_source, stem, + ipns_stem, + ipns_routing_url, rpc_url, ws_url, confirmation_depth, @@ -813,7 +838,8 @@ impl Commands { identity, listen, images, - } => Self::daemon_install(identity, listen, images, false) + ipns_routing_url, + } => Self::daemon_install(identity, listen, images, ipns_routing_url, false) .await .map(|_| ()), DaemonAction::Uninstall => Self::daemon_uninstall().await, @@ -1420,9 +1446,10 @@ wasip2::cli::command::export!({iface_name}Guest); identity: Option, listen: Vec, images: Vec, + ipns_routing_url: String, quiet: bool, ) -> Result { - daemon_cmd::daemon_install(identity, listen, images, quiet).await + daemon_cmd::daemon_install(identity, listen, images, ipns_routing_url, quiet).await } /// Remove the platform service file. @@ -1441,6 +1468,8 @@ wasip2::cli::command::export!({iface_name}Guest); wasm_debug: bool, kernel_source: ww::kernel::Source, stem: Option, + ipns_stem: Option, + ipns_routing_url: String, rpc_url: String, ws_url: String, confirmation_depth: u64, @@ -1470,7 +1499,7 @@ wasip2::cli::command::export!({iface_name}Guest); // Reject local configuration before Kubo readiness can intentionally // wait forever in production. A bad path must remain actionable. - if !mounts.is_empty() || stem.is_none() { + if !mounts.is_empty() || (stem.is_none() && ipns_stem.is_none()) { image::validate_mounts_virtual(&mounts)?; } let user_mounts = mounts.clone(); @@ -1628,12 +1657,96 @@ wasip2::cli::command::export!({iface_name}Guest); let mut all_mounts: Vec = Vec::new(); let stem_contract = stem.as_deref().map(parse_contract_address).transpose()?; + let ipns_stem_name = ipns_stem + .as_deref() + .map(ww::ipns::parse_name) + .transpose() + .context("parse --ipns-stem")?; + + // The installed `ww` namespace uses the host identity as its default + // IPNS signer. Publish before namespace resolution so a newly migrated + // name exists before Kubo resolves it. The republisher then becomes + // the only writer for this state directory in the running process. + let mut runtime_ns_configs = ns_configs.clone(); + if let Some(config) = ns_configs.iter().find(|config| config.name == "ww") { + let configured_name = if config.ipns.is_empty() { + None + } else { + // Namespace IPNS values may also be DNSLink names. Only a + // PeerID-form value can identify the host-owned publisher. + ww::ipns::parse_name(&config.ipns).ok() + }; + if configured_name == Some(peer_id) { + if let Some(path) = config.bootstrap_ipfs_path()? { + let desired: cid::Cid = path + .strip_prefix("/ipfs/") + .expect("validated bootstrap path has /ipfs prefix") + .parse() + .context("parse default ww namespace bootstrap CID")?; + let state_dir = dirs::home_dir() + .context("default IPNS publication requires a HOME directory")? + .join(".ww"); + let routing = ww::ipns::RoutingClient::new(ipns_routing_url.clone())?; + let routing_ready = match routing.probe().await { + Ok(()) => true, + Err(error) if error.kind() == ww::ipns::RoutingErrorKind::Unsupported => { + return Err(anyhow::Error::new(error).context( + "default IPNS publication requires Kubo HTTP Routing V1", + )); + } + Err(error) if error.kind() == ww::ipns::RoutingErrorKind::Temporary => { + tracing::warn!( + "Initial Routing V1 probe failed; default IPNS publication will retry: {error}" + ); + false + } + Err(error) => { + return Err(anyhow::Error::new(error) + .context("default IPNS Routing V1 probe failed")); + } + }; + let mut publisher = + ww::ipns::Publisher::open(keypair.clone(), routing, &state_dir)?; + let initial_publish_succeeded = if routing_ready { + match publisher.publish(desired).await { + Ok(()) => true, + Err(error) if ww::ipns::is_temporary_routing_failure(&error) => { + tracing::warn!( + "Initial host IPNS publication failed; republisher will retry: {error:#}" + ); + false + } + Err(error) => { + return Err(error).context("initial host IPNS publication failed"); + } + } + } else { + false + }; + if !initial_publish_succeeded { + if let Some(runtime_config) = runtime_ns_configs + .iter_mut() + .find(|runtime_config| runtime_config.name == config.name) + { + // Until the signed record is visible, select the + // immutable bootstrap instead of treating a 404 as + // a terminal namespace-resolution error. + runtime_config.ipns.clear(); + } + } + supervisor.try_spawn( + "ipns-republisher", + ww::ipns::Republisher::new(publisher, desired, initial_publish_succeeded), + )?; + } + } + } // Namespace layers sit between stem (on-chain base) and user mounts. - if !ns_configs.is_empty() { + if !runtime_ns_configs.is_empty() { let resolved = tokio::select! { resolved = ww::ns::resolve_namespaces( - &ns_configs, + &runtime_ns_configs, &boot_ipfs_client, &runtime_status, ) => resolved.context("Failed to resolve namespace configs")?, @@ -1829,16 +1942,35 @@ wasip2::cli::command::export!({iface_name}Guest); } None => (None, None), }; - let stem_source: Option> = stem_contract - .map(|contract| { - ww::stem::atom::Source::new(ww::stem::atom::Config::new( - rpc_url.clone(), - contract, - confirmation_depth, - )) - .map(|source| Box::new(source) as Box) - }) - .transpose()?; + let stem_source: Option> = if let Some(contract) = stem_contract { + Some(Box::new(ww::stem::atom::Source::new( + ww::stem::atom::Config::new(rpc_url.clone(), contract, confirmation_depth), + )?)) + } else if let Some(name) = ipns_stem_name { + let routing = ww::ipns::RoutingClient::new(ipns_routing_url.clone())?; + match routing.probe().await { + Ok(()) => {} + Err(error) if error.kind() == ww::ipns::RoutingErrorKind::Unsupported => { + return Err(anyhow::Error::new(error) + .context("--ipns-stem requires Kubo HTTP Routing V1")); + } + Err(error) if error.kind() == ww::ipns::RoutingErrorKind::Temporary => tracing::warn!( + "Initial IPNS Stem Routing V1 probe failed; Source.current() will retry: {error}" + ), + Err(error) => { + return Err(anyhow::Error::new(error) + .context("--ipns-stem Routing V1 probe failed")); + } + } + let state_dir = dirs::home_dir() + .context("--ipns-stem requires a HOME directory for durable watermark state")? + .join(".ww"); + Some(Box::new(ww::stem::ipns::Source::new( + ww::stem::ipns::Config::new(name, ipns_routing_url.clone(), state_dir), + )?)) + } else { + None + }; let deployment_config = ww::deployment::Config { source: stem_source, candidates: candidate_rx, @@ -2159,6 +2291,13 @@ wasip2::cli::command::export!({iface_name}Guest); if !ww_dir.exists() { bail!("~/.ww does not exist. Run `ww perform install` first."); } + let identity_path = ww_dir.join("identity"); + let identity_path_string = identity_path + .to_str() + .context("~/.ww/identity path is non-UTF-8")?; + let host_key = ww::keys::load(identity_path_string)?; + let host_peer_id = ww::keys::to_libp2p(&host_key)?.public().to_peer_id(); + let default_ipns_name = ww::ipns::canonical_name(host_peer_id); // Ensure subdirectories exist (may be missing if created by older version). for sub in &["bin", "etc/ns", "logs"] { @@ -2206,9 +2345,20 @@ wasip2::cli::command::export!({iface_name}Guest); skip("WASM images (unchanged)".into()); } - // ── Standard namespace republish (if images changed + Kubo running) ── + // ── Standard namespace indexing (if images changed + Kubo running) ── let ipfs_client = ipfs::HttpClient::new("http://localhost:5001".into()); let kubo_ok = ipfs_client.kubo_info().await.is_ok(); + if kubo_ok { + let routing_url = std::env::var("IPFS_ROUTING_API") + .unwrap_or_else(|_| "http://localhost:8080".to_string()); + ww::ipns::RoutingClient::new(routing_url)? + .probe() + .await + .context( + "Kubo HTTP Routing V1 is required for default IPNS publication; run `ipfs config --json Gateway.ExposeRoutingAPI true`, restart Kubo, and set IPFS_ROUTING_API when the Gateway is not http://localhost:8080", + )?; + } + let namespace_identity_changed; { let ns_path = ww_dir.join("etc/ns/ww"); @@ -2224,6 +2374,8 @@ wasip2::cli::command::export!({iface_name}Guest); bootstrap: std_cid.to_string(), } }; + namespace_identity_changed = config.ipns != default_ipns_name; + config.ipns.clone_from(&default_ipns_name); if kubo_ok && images_ok && any_images_changed { let sp = spin(); @@ -2246,9 +2398,6 @@ wasip2::cli::command::export!({iface_name}Guest); let ipfs_path = format!("/ipfs/{cid}"); config.bootstrap = ipfs_path.clone(); let _ = ipfs_client.pin_add(&ipfs_path).await; - if !config.ipns.is_empty() { - let _ = ipfs_client.name_publish(&ipfs_path, "ww").await; - } sp.finish_and_clear(); done(format!("Standard namespace ({ipfs_path})")); } @@ -2275,16 +2424,27 @@ wasip2::cli::command::export!({iface_name}Guest); } // ── Install-owned FHS + daemon service file (unconditional) ── - let identity_path = ww_dir.join("identity"); let (_, install_layer_changed) = daemon_cmd::refresh_default_fhs(&ww_dir)?; // Empty images select ~/.ww/fhs and read ~/.ww/etc/ns through the // separate host-only namespace root. - let service_definition_changed = - Self::daemon_install(Some(identity_path), Vec::new(), Vec::new(), true).await?; + let ipns_routing_url = std::env::var("IPFS_ROUTING_API") + .unwrap_or_else(|_| "http://localhost:8080".to_string()); + let service_definition_changed = Self::daemon_install( + Some(identity_path), + Vec::new(), + Vec::new(), + ipns_routing_url, + true, + ) + .await?; done("Background daemon".into()); // ── Restart daemon when deployed content or its boundary changed ─ - if !any_images_changed && !install_layer_changed && !service_definition_changed { + if !any_images_changed + && !install_layer_changed + && !service_definition_changed + && !namespace_identity_changed + { skip("Daemon restart (nothing changed)".into()); } else { match Self::restart_user_daemon(&home) { @@ -2292,15 +2452,19 @@ wasip2::cli::command::export!({iface_name}Guest); Some(false) => { if cfg!(target_os = "macos") { let plist_path = home.join("Library/LaunchAgents/io.wetware.ww.plist"); - fail(format!( - "Daemon start (try: launchctl load {})", + bail!( + "updated namespace was not published because the daemon could not restart; try: launchctl load {}", plist_path.display() - )); + ); } else { - fail("Daemon restart (try: systemctl --user restart ww)".into()); + bail!( + "updated namespace was not published because the daemon could not restart; try: systemctl --user restart ww" + ); } } - None => skip("Daemon start (no service file)".into()), + None => bail!( + "updated namespace was not published because no daemon service file was available" + ), } } @@ -2357,52 +2521,10 @@ wasip2::cli::command::export!({iface_name}Guest); let kp = ww::keys::to_libp2p(&sk)?; let peer_id = kp.public().to_peer_id(); done(format!("Identity ({peer_id})")); - - // ── IPNS key (first-time only, before update so publish works) ─ - let ipfs_client = ipfs::HttpClient::new("http://localhost:5001".into()); - let kubo_ok = ipfs_client.kubo_info().await.is_ok(); - - if kubo_ok { - use indicatif::{ProgressBar, ProgressStyle}; - use std::time::Duration; - - let spin = || { - let pb = ProgressBar::new_spinner(); - pb.set_style( - ProgressStyle::default_spinner() - .template(" \u{2699} {msg}") - .expect("valid template"), - ); - pb.enable_steady_tick(Duration::from_millis(80)); - pb - }; - let fail = |msg: String| println!(" \u{2717} {msg}"); - - let keys = ipfs_client.key_list().await.unwrap_or_default(); - if !keys.iter().any(|k| k == "ww") { - let sp = spin(); - sp.set_message("Generating IPNS key..."); - match ipfs_client.key_gen("ww").await { - Ok(id) => { - // Write the key into namespace config so perform_update - // can publish to IPNS on the first install. - let ns_path = ww_dir.join("etc/ns/ww"); - let config = ww::ns::NamespaceConfig { - name: "ww".to_string(), - ipns: id.clone(), - bootstrap: ww::namespace::WW_STD_CID.to_string(), - }; - let _ = config.write_to(&ns_path); - sp.finish_and_clear(); - done(format!("IPNS key ({id})")); - } - Err(e) => { - sp.finish_and_clear(); - fail(format!("IPNS key ({e})")); - } - } - } - } + done(format!( + "Default IPNS name ({})", + ww::ipns::canonical_name(peer_id) + )); // ── Update: WASM images, stdlib, daemon ───────────────────── Self::perform_update().await?; @@ -2926,6 +3048,8 @@ mod tests { identity: None, insecure_ephemeral: false, stem: None, + ipns_stem: None, + ipns_routing_url: "http://localhost:8080".to_string(), rpc_url: "http://127.0.0.1:8545".to_string(), ws_url: "ws://127.0.0.1:8545".to_string(), confirmation_depth: 6, @@ -2971,6 +3095,21 @@ mod tests { assert_eq!(parse_run_admin_addr(&["ww", "run"]), "127.0.0.1:2026"); } + #[test] + fn atom_and_ipns_stems_are_mutually_exclusive() { + let error = Cli::try_parse_from([ + "ww", + "run", + "--stem", + "0x0000000000000000000000000000000000000000", + "--ipns-stem", + "12D3KooWGuR5BdSqp23UeoeesuwYwW3ebQ9rZ8aVwfWEDU8kvCYJ", + ]) + .err() + .expect("Atom and IPNS Stem options must conflict"); + assert_eq!(error.kind(), clap::error::ErrorKind::ArgumentConflict); + } + #[test] fn run_admin_off_disables_case_insensitively() { for value in ["off", "OFF", " Off "] { @@ -3164,6 +3303,7 @@ mod tests { namespace_roots: vec![PathBuf::from("/tmp/host-state")], images: Vec::new(), http_listen: Some(addr.to_string()), + ipns_routing_url: "http://localhost:8080".to_string(), } } @@ -3174,6 +3314,7 @@ mod tests { namespace_roots: Vec::new(), images: Vec::new(), http_listen: None, + ipns_routing_url: "http://localhost:8080".to_string(), } } @@ -3210,6 +3351,11 @@ mod tests { && plist.contains("/tmp/host-state"), "plist should read namespace configuration through a host-only root, got:\n{plist}" ); + assert!( + plist.contains("--ipns-routing-url") + && plist.contains("http://localhost:8080"), + "plist should preserve the Routing V1 listener, got:\n{plist}" + ); } #[test] @@ -3258,6 +3404,10 @@ mod tests { unit.contains("--namespace-root /tmp/host-state"), "systemd unit should read namespace configuration through a host-only root, got:\n{unit}" ); + assert!( + unit.contains("--ipns-routing-url http://localhost:8080"), + "systemd unit should preserve the Routing V1 listener, got:\n{unit}" + ); } #[test] diff --git a/src/deployment/mod.rs b/src/deployment/mod.rs index 710ce3bc..a11eda0a 100644 --- a/src/deployment/mod.rs +++ b/src/deployment/mod.rs @@ -2870,6 +2870,124 @@ mod tests { server.await.unwrap(); } + #[tokio::test] + async fn ipns_eol_revokes_generation_and_real_source_recovers_host() { + use chrono::Utc; + use libp2p::identity::Keypair; + use rust_ipns::Record; + + let state = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let root: cid::Cid = ROOT.parse().unwrap(); + let current = Record::new( + &keypair, + format!("/ipfs/{root}"), + Utc::now() + chrono::Duration::seconds(1), + 1, + Duration::from_millis(1), + ) + .unwrap() + .encode() + .unwrap(); + crate::ipns::RecordStore::follower(state.path(), name) + .persist(¤t) + .unwrap(); + let recovery = Record::new( + &keypair, + format!("/ipfs/{root}"), + Utc::now() + chrono::Duration::hours(1), + 2, + Duration::from_millis(1), + ) + .unwrap() + .encode() + .unwrap(); + + let routing_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let routing_address = routing_listener.local_addr().unwrap(); + let routing_server = tokio::spawn(async move { + for (status, body) in [ + ("400 Bad Request", Vec::new()), + ("503 Service Unavailable", Vec::new()), + ("200 OK", recovery), + ] { + let (mut stream, _) = routing_listener.accept().await.unwrap(); + let _request = read_request(&mut stream).await; + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + crate::ipns::IPNS_RECORD_MEDIA_TYPE, + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + stream.write_all(&body).await.unwrap(); + } + }); + let source = crate::stem::ipns::Source::new(crate::stem::ipns::Config::new( + name, + format!("http://{routing_address}"), + state.path().to_path_buf(), + )) + .unwrap(); + let (ipfs_client, _calls, _requests, shutdown_tx, ipfs_server) = recording_kubo().await; + let (epoch_tx, epoch_rx) = watch::channel(Epoch::zero()); + let staging = tempfile::tempdir().unwrap(); + let mut deployment = Deployment::bootstrap( + Config { + source: Some(Box::new(source)), + candidates: None, + frozen_layers: Vec::new(), + ipfs_client, + staging_dir: staging.path().to_path_buf(), + }, + epoch_tx, + epoch_rx, + ) + .await + .unwrap(); + assert_eq!(deployment.current_epoch().root.as_deref(), Some(ROOT)); + let mut epoch_observer = deployment.epoch_rx.clone(); + epoch_observer.borrow_and_update(); + + let (running, result_tx, mut terminate_rx) = test_running_generation(); + let outcome = { + let transition = deployment.await_generation(running); + tokio::pin!(transition); + tokio::select! { + _ = &mut transition => panic!("IPNS transition completed before revocation"), + changed = epoch_observer.changed() => changed.unwrap(), + _ = tokio::time::sleep(Duration::from_secs(3)) => { + panic!("IPNS EOL did not revoke deployment") + } + } + assert_eq!(epoch_observer.borrow().seq, 1); + assert_eq!(epoch_observer.borrow().root, None); + terminate_rx.changed().await.unwrap(); + result_tx.send(Ok(kernel::Outcome::Terminated)).unwrap(); + tokio::time::timeout(Duration::from_secs(3), transition) + .await + .expect("IPNS recovery did not reactivate deployment") + .unwrap() + }; + + assert!(matches!( + outcome, + Outcome::Replaced { + old_epoch: 0, + new_epoch: 2, + .. + } + )); + assert_eq!(deployment.current_epoch().root.as_deref(), Some(ROOT)); + assert!( + deployment.source_task.is_some(), + "host follower must remain alive" + ); + routing_server.await.unwrap(); + shutdown_tx.send(()).unwrap(); + ipfs_server.await.unwrap(); + } + #[tokio::test] async fn source_error_does_not_advance_the_epoch() { let client = crate::ipfs::HttpClient::new("http://127.0.0.1:1".to_owned()); diff --git a/src/ipns.rs b/src/ipns.rs new file mode 100644 index 00000000..d30541a2 --- /dev/null +++ b/src/ipns.rs @@ -0,0 +1,1279 @@ +//! Host-owned IPNS record handling. +//! +//! Wetware signs and validates records locally. Kubo only transports the raw +//! signed protobuf through HTTP Routing V1. The canonical durable object is +//! always that raw signed record. + +use std::cmp::Ordering; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{bail, Context, Result}; +use chrono::{DateTime, Utc}; +use cid::multibase::Base; +use cid::Cid; +use libp2p::identity::{Keypair, PeerId}; +use reqwest::StatusCode; +use rust_ipns::Record; + +pub const IPNS_RECORD_MEDIA_TYPE: &str = "application/vnd.ipfs.ipns-record"; +pub const MAX_RECORD_SIZE: usize = 10_240; + +/// Kubo/Boxo-compatible defaults used by the host-owned publisher. +pub const RECORD_LIFETIME: Duration = Duration::from_secs(48 * 60 * 60); +pub const RECORD_TTL: Duration = Duration::from_secs(5 * 60); +pub const REPUBLISH_INTERVAL: Duration = Duration::from_secs(4 * 60 * 60); +pub const INITIAL_REPUBLISH_DELAY: Duration = Duration::from_secs(60); +pub const REPUBLISH_RETRY_DELAY: Duration = Duration::from_secs(5 * 60); +pub const ROUTING_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + +const LIBP2P_KEY_CODEC: u64 = 0x72; + +/// Parse either a canonical Base36 IPNS name or a legacy base58 Peer ID. +pub fn parse_name(value: &str) -> Result { + let value = value.strip_prefix("/ipns/").unwrap_or(value); + if let Ok(peer_id) = value.parse::() { + return Ok(peer_id); + } + let cid = value + .parse::() + .with_context(|| format!("invalid IPNS name: {value}"))?; + if cid.codec() != LIBP2P_KEY_CODEC { + bail!( + "invalid IPNS name codec 0x{:x}; expected libp2p-key (0x72)", + cid.codec() + ); + } + PeerId::from_multihash(cid.hash().to_owned()) + .map_err(|_| anyhow::anyhow!("IPNS name contains an invalid Peer ID multihash")) +} + +/// Format a Peer ID as the canonical Base36 CIDv1 IPNS name. +pub fn canonical_name(peer_id: PeerId) -> String { + Cid::new_v1(LIBP2P_KEY_CODEC, peer_id.into()) + .to_string_of_base(Base::Base36Lower) + .expect("Base36 is a valid CID multibase") +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RoutingErrorKind { + Unsupported, + Temporary, + Malformed, + Rejected, +} + +#[derive(Debug)] +pub struct RoutingError { + kind: RoutingErrorKind, + message: String, +} + +impl RoutingError { + fn new(kind: RoutingErrorKind, message: impl Into) -> Self { + Self { + kind, + message: message.into(), + } + } + + pub fn kind(&self) -> RoutingErrorKind { + self.kind + } +} + +impl std::fmt::Display for RoutingError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.message) + } +} + +impl std::error::Error for RoutingError {} + +/// Return true when an operation failed only because Routing V1 is temporarily +/// unavailable. Local state, signing, and protocol-policy failures are fatal. +pub fn is_temporary_routing_failure(error: &anyhow::Error) -> bool { + error + .chain() + .find_map(|cause| cause.downcast_ref::()) + .is_some_and(|error| error.kind() == RoutingErrorKind::Temporary) +} + +#[derive(Debug)] +pub enum Fetch { + Record(Vec), + NotFound, +} + +/// Narrow HTTP Routing V1 raw-record client. +#[derive(Clone)] +pub struct RoutingClient { + base_url: String, + http: reqwest::Client, +} + +impl RoutingClient { + pub fn new(base_url: impl Into) -> Result { + Self::with_timeout(base_url, ROUTING_REQUEST_TIMEOUT) + } + + fn with_timeout(base_url: impl Into, timeout: Duration) -> Result { + let http = reqwest::Client::builder() + .no_proxy() + .timeout(timeout) + .build() + .context("build Routing V1 HTTP client")?; + Ok(Self { + base_url: base_url.into().trim_end_matches('/').to_string(), + http, + }) + } + + fn record_url(&self, name: PeerId) -> String { + format!("{}/routing/v1/ipns/{}", self.base_url, canonical_name(name)) + } + + /// Verify that the configured listener exposes Routing V1. + /// + /// Kubo returns 400 for a malformed IPNS name when the route exists and + /// 404 from the ordinary Gateway when `Gateway.ExposeRoutingAPI` is false. + pub async fn probe(&self) -> std::result::Result<(), RoutingError> { + let response = self + .http + .get(format!( + "{}/routing/v1/ipns/not-an-ipns-name", + self.base_url + )) + .header(reqwest::header::ACCEPT, IPNS_RECORD_MEDIA_TYPE) + .send() + .await + .map_err(|error| { + RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 probe transport failure: {error}"), + ) + })?; + match response.status() { + StatusCode::BAD_REQUEST => Ok(()), + StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED => { + Err(RoutingError::new( + RoutingErrorKind::Unsupported, + "Routing V1 endpoint is unavailable; set Kubo Gateway.ExposeRoutingAPI=true and configure the Gateway listener, not the port 5001 RPC listener", + )) + } + status + if status.is_server_error() + || matches!(status, StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS) => + { + Err(RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 probe returned {status}"), + )) + } + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => Err(RoutingError::new( + RoutingErrorKind::Rejected, + format!("Routing V1 probe rejected access with {}", response.status()), + )), + status => Err(RoutingError::new( + RoutingErrorKind::Malformed, + format!("Routing V1 probe returned unexpected status {status}"), + )), + } + } + + pub async fn get(&self, name: PeerId) -> std::result::Result { + let mut response = self + .http + .get(self.record_url(name)) + .header(reqwest::header::ACCEPT, IPNS_RECORD_MEDIA_TYPE) + .send() + .await + .map_err(|error| { + RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 GET transport failure: {error}"), + ) + })?; + let status = response.status(); + if status == StatusCode::NOT_FOUND { + return Ok(Fetch::NotFound); + } + if matches!( + status, + StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED + ) { + return Err(RoutingError::new( + RoutingErrorKind::Unsupported, + "Routing V1 endpoint does not support raw IPNS GET; enable Kubo Gateway.ExposeRoutingAPI", + )); + } + if status.is_server_error() + || matches!( + status, + StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS + ) + { + return Err(RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 GET returned {status}"), + )); + } + if !status.is_success() { + return Err(RoutingError::new( + RoutingErrorKind::Rejected, + format!("Routing V1 GET returned {status}"), + )); + } + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + let media_type = content_type + .split(';') + .next() + .map(str::trim) + .unwrap_or_default(); + if media_type.eq_ignore_ascii_case("text/plain") { + return Ok(Fetch::NotFound); + } + if !content_type + .to_ascii_lowercase() + .starts_with(IPNS_RECORD_MEDIA_TYPE) + { + return Err(RoutingError::new( + RoutingErrorKind::Malformed, + format!("Routing V1 GET returned unexpected media type {content_type:?}"), + )); + } + if response + .content_length() + .is_some_and(|length| length > MAX_RECORD_SIZE as u64) + { + return Err(RoutingError::new( + RoutingErrorKind::Malformed, + "Routing V1 GET record exceeds 10,240 bytes", + )); + } + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(|error| { + RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 GET response body failed: {error}"), + ) + })? { + if bytes.len() + chunk.len() > MAX_RECORD_SIZE { + return Err(RoutingError::new( + RoutingErrorKind::Malformed, + "Routing V1 GET record exceeds 10,240 bytes", + )); + } + bytes.extend_from_slice(&chunk); + } + Ok(Fetch::Record(bytes)) + } + + pub async fn put(&self, name: PeerId, raw: &[u8]) -> std::result::Result<(), RoutingError> { + if raw.len() > MAX_RECORD_SIZE { + return Err(RoutingError::new( + RoutingErrorKind::Malformed, + "refusing to PUT an IPNS record larger than 10,240 bytes", + )); + } + let response = self + .http + .put(self.record_url(name)) + .header(reqwest::header::CONTENT_TYPE, IPNS_RECORD_MEDIA_TYPE) + .body(raw.to_vec()) + .send() + .await + .map_err(|error| { + RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 PUT transport failure: {error}"), + ) + })?; + let status = response.status(); + if matches!( + status, + StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED + ) { + return Err(RoutingError::new( + RoutingErrorKind::Unsupported, + "Routing V1 endpoint is unavailable; enable Kubo Gateway.ExposeRoutingAPI and use the Gateway listener", + )); + } + if status.is_server_error() + || matches!( + status, + StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS + ) + { + return Err(RoutingError::new( + RoutingErrorKind::Temporary, + format!("Routing V1 PUT returned {status}"), + )); + } + if !status.is_success() { + let detail = response.text().await.unwrap_or_default(); + let detail = detail.trim(); + return Err(RoutingError::new( + RoutingErrorKind::Rejected, + if detail.is_empty() { + format!("Routing V1 PUT rejected the record with {status}") + } else { + format!("Routing V1 PUT rejected the record with {status}: {detail}") + }, + )); + } + Ok(()) + } +} + +/// One durable raw-record location below trusted private host state. +#[derive(Clone, Debug)] +pub struct RecordStore { + state_dir: PathBuf, + path: PathBuf, +} + +impl RecordStore { + pub fn follower(state_dir: &Path, name: PeerId) -> Self { + Self::new(state_dir, "follow", name) + } + + pub fn publisher(state_dir: &Path, name: PeerId) -> Self { + Self::new(state_dir, "publish", name) + } + + fn new(state_dir: &Path, role: &str, name: PeerId) -> Self { + let name = canonical_name(name); + Self { + state_dir: state_dir.to_path_buf(), + path: state_dir + .join("ipns") + .join(role) + .join(format!("{name}.record")), + } + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn load(&self) -> Result>> { + match std::fs::read(&self.path) { + Ok(bytes) if bytes.len() <= MAX_RECORD_SIZE => Ok(Some(bytes)), + Ok(_) => bail!( + "persisted IPNS record exceeds 10,240 bytes: {}", + self.path.display() + ), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error) + .with_context(|| format!("read persisted IPNS record: {}", self.path.display())), + } + } + + pub fn persist(&self, raw: &[u8]) -> Result<()> { + if raw.len() > MAX_RECORD_SIZE { + bail!("refusing to persist an IPNS record larger than 10,240 bytes"); + } + self.restrict_directories()?; + crate::keys::atomic_write_private(&self.path, raw) + .with_context(|| format!("persist signed IPNS record: {}", self.path.display())) + } + + fn restrict_directories(&self) -> Result<()> { + #[cfg(unix)] + use std::os::unix::fs::PermissionsExt; + + for directory in [ + self.state_dir.clone(), + self.state_dir.join("ipns"), + self.path + .parent() + .expect("record path always has a parent") + .to_path_buf(), + ] { + std::fs::create_dir_all(&directory).with_context(|| { + format!( + "create private IPNS state directory: {}", + directory.display() + ) + })?; + #[cfg(unix)] + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o700)) + .with_context(|| { + format!( + "restrict private IPNS state directory: {}", + directory.display() + ) + })?; + } + Ok(()) + } +} + +/// A locally verified signed record and its authenticated semantics. +#[derive(Clone, Debug)] +pub struct SignedRecord { + raw: Vec, + record: Record, + value: Vec, + deployment_cid: Option, + eol: DateTime, + ttl: Duration, +} + +impl SignedRecord { + pub fn decode(name: PeerId, raw: Vec) -> Result { + let record = Record::decode(&raw).context("decode raw IPNS record")?; + record + .verify_signature(name) + .context("verify IPNS record signer and signature")?; + let value = record.value().to_vec(); + let deployment_cid = parse_value(&value).ok(); + let eol = record + .validity() + .context("parse signed IPNS record EOL")? + .with_timezone(&Utc); + let ttl = Duration::from_nanos(record.ttl()); + Ok(Self { + raw, + record, + value, + deployment_cid, + eol, + ttl, + }) + } + + pub fn raw(&self) -> &[u8] { + &self.raw + } + + pub fn value(&self) -> &[u8] { + &self.value + } + + pub fn deployment_cid(&self) -> Option { + self.deployment_cid + } + + pub fn sequence(&self) -> u64 { + self.record.sequence() + } + + pub fn eol(&self) -> DateTime { + self.eol + } + + pub fn ttl(&self) -> Duration { + self.ttl + } + + pub fn is_expired_at(&self, now: DateTime) -> bool { + self.eol <= now + } + + pub fn compare(&self, other: &Self) -> Result { + self.record + .compare(&other.record) + .context("compare signed IPNS records") + } +} + +fn parse_value(value: &[u8]) -> Result { + let value = std::str::from_utf8(value).context("IPNS value is not UTF-8")?; + let cid = value + .strip_prefix("/ipfs/") + .context("IPNS value is not an /ipfs/ deployment binding")?; + if cid.is_empty() || cid.contains('/') { + bail!("IPNS deployment binding must contain one CID and no subpath"); + } + cid.parse::() + .context("IPNS value contains an invalid CID") +} + +/// Apply the Wetware single-writer publisher policy to durable local state. +pub struct Publisher { + keypair: Keypair, + name: PeerId, + routing: RoutingClient, + store: RecordStore, + local: Option, + pending_put: bool, +} + +/// Long-running publisher lifecycle for the host's own default IPNS name. +/// +/// A followed third-party name never creates this service. +pub struct Republisher { + publisher: Publisher, + desired: Cid, + initial_publish_succeeded: bool, +} + +impl Republisher { + pub fn new(publisher: Publisher, desired: Cid, initial_publish_succeeded: bool) -> Self { + Self { + publisher, + desired, + initial_publish_succeeded, + } + } + + async fn wait_or_shutdown( + shutdown: &mut tokio::sync::watch::Receiver<()>, + duration: Duration, + ) -> bool { + tokio::select! { + _ = tokio::time::sleep(duration) => false, + _ = shutdown.changed() => true, + } + } + + async fn run_async(mut self, mut shutdown: tokio::sync::watch::Receiver<()>) -> Result<()> { + let mut retry_delay = !self.initial_publish_succeeded; + let mut first = true; + loop { + let delay = if retry_delay { + REPUBLISH_RETRY_DELAY + } else if first { + INITIAL_REPUBLISH_DELAY + } else { + REPUBLISH_INTERVAL + }; + if Self::wait_or_shutdown(&mut shutdown, delay).await { + return Ok(()); + } + + let result = if self.publisher.has_pending_put() { + self.publisher.retry_pending(self.desired).await + } else if retry_delay { + self.publisher.publish(self.desired).await + } else { + self.publisher.refresh(self.desired).await + }; + match result { + Ok(()) => { + retry_delay = false; + first = false; + tracing::info!( + ipns_name = %self.publisher.canonical_name(), + sequence = self.publisher.current().map(SignedRecord::sequence), + "Host IPNS record published" + ); + } + Err(error) => { + if !is_temporary_routing_failure(&error) { + return Err(error) + .context("host IPNS republisher stopped on a non-retryable failure"); + } + retry_delay = true; + tracing::warn!( + ipns_name = %self.publisher.canonical_name(), + retry_secs = REPUBLISH_RETRY_DELAY.as_secs(), + pending_put = self.publisher.has_pending_put(), + "Host IPNS publication failed temporarily; publication will be retried: {error:#}" + ); + } + } + } + } +} + +impl crate::services::Service for Republisher { + fn run(self, shutdown: tokio::sync::watch::Receiver<()>) -> Result<()> { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .context("build IPNS republisher runtime")? + .block_on(self.run_async(shutdown)) + } +} + +impl Publisher { + pub fn open(keypair: Keypair, routing: RoutingClient, state_dir: &Path) -> Result { + let name = keypair.public().to_peer_id(); + let store = RecordStore::publisher(state_dir, name); + let local = store + .load()? + .map(|raw| SignedRecord::decode(name, raw)) + .transpose() + .context("validate persisted publisher record")?; + Ok(Self { + keypair, + name, + routing, + store, + local, + pending_put: false, + }) + } + + pub fn name(&self) -> PeerId { + self.name + } + + pub fn canonical_name(&self) -> String { + canonical_name(self.name) + } + + pub fn state_path(&self) -> &Path { + self.store.path() + } + + pub fn current(&self) -> Option<&SignedRecord> { + self.local.as_ref() + } + + /// Publish a desired deployment CID. + /// + /// An already-signed matching record is retried byte-for-byte. Creating a + /// changed binding first reconciles the valid signed network floor. + pub async fn publish(&mut self, value: Cid) -> Result<()> { + self.reconcile_network(value).await?; + if let Some(local) = self.local.as_ref() { + if local.deployment_cid() == Some(value) && !local.is_expired_at(Utc::now()) { + let refresh_deadline = Utc::now() + + chrono::Duration::from_std(INITIAL_REPUBLISH_DELAY) + .expect("initial republish delay fits chrono"); + if local.eol() > refresh_deadline { + return self.put_current().await; + } + return self.refresh_after_reconcile(value).await; + } + } + + let sequence = match self.local.as_ref() { + Some(local) if local.deployment_cid() == Some(value) => local.sequence(), + Some(local) => local + .sequence() + .checked_add(1) + .context("IPNS publisher sequence exhausted")?, + None => 0, + }; + self.sign_persist_and_put(value, sequence).await + } + + /// Reconcile and extend the desired binding without changing its sequence. + pub async fn refresh(&mut self, desired: Cid) -> Result<()> { + self.reconcile_network(desired).await?; + self.refresh_after_reconcile(desired).await + } + + async fn refresh_after_reconcile(&mut self, desired: Cid) -> Result<()> { + let current = self + .local + .as_ref() + .context("cannot republish before an IPNS value has been published")?; + if current.deployment_cid() != Some(desired) { + let sequence = current + .sequence() + .checked_add(1) + .context("IPNS publisher sequence exhausted")?; + return self.sign_persist_and_put(desired, sequence).await; + } + + let candidate = self.sign_record(desired, current.sequence())?; + if candidate.compare(current)? != Ordering::Greater { + return self.put_current().await; + } + self.persist_and_put(candidate).await + } + + fn has_pending_put(&self) -> bool { + self.pending_put + } + + /// Retry bytes only when this process persisted them before a failed PUT. + async fn retry_pending(&mut self, desired: Cid) -> Result<()> { + if !self.pending_put { + return self.publish(desired).await; + } + if self + .local + .as_ref() + .is_none_or(|record| record.is_expired_at(Utc::now())) + { + self.pending_put = false; + return self.publish(desired).await; + } + self.put_current().await + } + + async fn put_current(&mut self) -> Result<()> { + let current = self + .local + .as_ref() + .context("cannot retry before an IPNS record has been persisted")?; + let result = self + .routing + .put(self.name, current.raw()) + .await + .context("publish persisted IPNS record"); + if result.is_ok() { + self.pending_put = false; + } + result + } + + async fn sign_persist_and_put(&mut self, value: Cid, sequence: u64) -> Result<()> { + let signed = self.sign_record(value, sequence)?; + self.persist_and_put(signed).await + } + + fn sign_record(&self, value: Cid, sequence: u64) -> Result { + let eol = Utc::now() + + chrono::Duration::from_std(RECORD_LIFETIME) + .expect("48-hour record lifetime fits chrono"); + let record = Record::new( + &self.keypair, + format!("/ipfs/{value}").as_bytes(), + eol, + sequence, + RECORD_TTL, + ) + .context("sign IPNS record")?; + let raw = record.encode().context("encode signed IPNS record")?; + SignedRecord::decode(self.name, raw) + } + + async fn persist_and_put(&mut self, signed: SignedRecord) -> Result<()> { + // Persist-before-PUT also makes a failed PUT retryable byte-for-byte. + self.store.persist(signed.raw())?; + self.local = Some(signed); + self.pending_put = true; + self.put_current().await + } + + async fn reconcile_network(&mut self, desired: Cid) -> Result<()> { + let remote = match self.routing.get(self.name).await? { + Fetch::NotFound => return Ok(()), + Fetch::Record(raw) => SignedRecord::decode(self.name, raw) + .context("validate network IPNS publisher record")?, + }; + let Some(local) = self.local.as_ref() else { + self.store.persist(remote.raw())?; + let conflict = remote.deployment_cid() != Some(desired); + self.local = Some(remote); + self.pending_put = false; + if conflict { + return self.unexpected_remote_value(desired); + } + return Ok(()); + }; + + if local.sequence() == remote.sequence() && local.value() != remote.value() { + bail!( + "IPNS publisher equivocation at sequence {} for {}; another writer is using this private identity", + local.sequence(), + canonical_name(self.name) + ); + } + if remote.compare(local)? == Ordering::Greater { + let conflict = remote.deployment_cid() != Some(desired); + self.store.persist(remote.raw())?; + self.local = Some(remote); + self.pending_put = false; + if conflict { + return self.unexpected_remote_value(desired); + } + } + Ok(()) + } + + fn unexpected_remote_value(&self, desired: Cid) -> Result<()> { + let remote = self + .local + .as_ref() + .expect("network reconciliation persisted the remote floor"); + bail!( + "IPNS publisher single-writer conflict for {}: adopted network sequence {} selecting {:?}, but local configuration selects /ipfs/{desired}; no record was published", + canonical_name(self.name), + remote.sequence(), + String::from_utf8_lossy(remote.value()) + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + #[derive(Debug)] + struct Request { + method: String, + body: Vec, + } + + async fn read_request(stream: &mut tokio::net::TcpStream) -> Request { + let mut bytes = Vec::new(); + let header_end = loop { + let mut chunk = [0_u8; 4096]; + let read = stream.read(&mut chunk).await.unwrap(); + assert!(read > 0, "connection closed before HTTP headers"); + bytes.extend_from_slice(&chunk[..read]); + if let Some(offset) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { + break offset + 4; + } + }; + let headers = String::from_utf8_lossy(&bytes[..header_end]); + let method = headers.split_whitespace().next().unwrap().to_string(); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .unwrap_or(0); + while bytes.len() - header_end < content_length { + let mut chunk = [0_u8; 4096]; + let read = stream.read(&mut chunk).await.unwrap(); + assert!(read > 0, "connection closed before HTTP body"); + bytes.extend_from_slice(&chunk[..read]); + } + Request { + method, + body: bytes[header_end..header_end + content_length].to_vec(), + } + } + + async fn scripted_server( + responses: Vec<(&'static str, Vec)>, + ) -> (String, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { + let mut requests = Vec::new(); + for (status, body) in responses { + let (mut stream, _) = listener.accept().await.unwrap(); + requests.push(read_request(&mut stream).await); + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: {IPNS_RECORD_MEDIA_TYPE}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + stream.write_all(&body).await.unwrap(); + } + requests + }); + (format!("http://{address}"), task) + } + + fn test_cid(byte: u8) -> Cid { + use cid::multihash::Multihash; + Cid::new_v1(0x55, Multihash::<64>::wrap(0x00, &[byte]).unwrap()) + } + + #[test] + fn canonical_name_roundtrip() { + let keypair = Keypair::generate_ed25519(); + let peer_id = keypair.public().to_peer_id(); + let name = canonical_name(peer_id); + assert!(name.starts_with("k51")); + assert_eq!(parse_name(&name).unwrap(), peer_id); + assert_eq!(parse_name(&peer_id.to_string()).unwrap(), peer_id); + } + + #[test] + fn signed_record_roundtrip_uses_authenticated_fields() { + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let cid = test_cid(1); + let record = Record::new( + &keypair, + format!("/ipfs/{cid}"), + Utc::now() + chrono::Duration::hours(1), + 7, + RECORD_TTL, + ) + .unwrap(); + let decoded = SignedRecord::decode(name, record.encode().unwrap()).unwrap(); + assert_eq!(decoded.deployment_cid(), Some(cid)); + assert_eq!(decoded.sequence(), 7); + assert_eq!(decoded.ttl(), RECORD_TTL); + } + + #[test] + fn official_v2_only_record_is_consumed() { + // IPNS Record specification vector 6. The record omits every legacy + // V1 field and therefore exercises the reviewed PR #503 behavior. + let raw = hex::decode(concat!( + "42406a7698585e2b170709d4947e148b18b120d76db43515edacf290df96b71e", + "29c0ca5a16510256814f8255532ec4549ade26f4ca8335f4c41e2cfffeb1ba0", + "aa5014a78a56354544c1b000001a3185c50006556616c756558242f697066732f", + "6261666b7161647477676977773633746d70657168657a6c646e357a67696853", + "657175656e6365006856616c6964697479581b323132332d30382d3134543132", + "3a31373a30332e3639343035325a6c56616c69646974795479706500" + )) + .unwrap(); + let name: PeerId = "12D3KooWGuR5BdSqp23UeoeesuwYwW3ebQ9rZ8aVwfWEDU8kvCYJ" + .parse() + .unwrap(); + let record = SignedRecord::decode(name, raw).unwrap(); + assert_eq!(record.sequence(), 0); + assert_eq!(record.value(), b"/ipfs/bafkqadtwgiww63tmpeqhezldn5zgi"); + assert!(record.deployment_cid().is_some()); + } + + #[cfg(unix)] + #[test] + fn record_store_is_private_and_outside_fhs() { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let store = RecordStore::publisher(directory.path(), keypair.public().to_peer_id()); + store.persist(b"signed-record").unwrap(); + + assert!(!store.path().starts_with(directory.path().join("fhs"))); + assert_eq!( + std::fs::metadata(store.path()) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o600 + ); + assert_eq!( + std::fs::metadata(store.path().parent().unwrap()) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o700 + ); + assert_eq!(store.load().unwrap().unwrap(), b"signed-record"); + } + + #[tokio::test] + async fn routing_get_rejects_oversized_record() { + let (url, server) = + scripted_server(vec![("200 OK", vec![0_u8; MAX_RECORD_SIZE + 1])]).await; + let client = RoutingClient::new(url).unwrap(); + let error = client + .get(Keypair::generate_ed25519().public().to_peer_id()) + .await + .unwrap_err(); + assert_eq!(error.kind(), RoutingErrorKind::Malformed); + server.await.unwrap(); + } + + #[tokio::test] + async fn routing_get_accepts_current_and_newer_missing_record_representations() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + for (status, content_type, body) in [ + ("404 Not Found", IPNS_RECORD_MEDIA_TYPE, b"".as_slice()), + ( + "200 OK", + "text/plain; charset=utf-8", + b"not found".as_slice(), + ), + ("200 OK", "application/json", b"{}".as_slice()), + ] { + let (mut stream, _) = listener.accept().await.unwrap(); + let _request = read_request(&mut stream).await; + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + stream.write_all(body).await.unwrap(); + } + }); + let client = RoutingClient::new(format!("http://{address}")).unwrap(); + let name = Keypair::generate_ed25519().public().to_peer_id(); + + assert!(matches!(client.get(name).await.unwrap(), Fetch::NotFound)); + assert!(matches!(client.get(name).await.unwrap(), Fetch::NotFound)); + let error = client.get(name).await.unwrap_err(); + assert_eq!(error.kind(), RoutingErrorKind::Malformed); + server.await.unwrap(); + } + + #[tokio::test] + async fn stalled_routing_request_times_out_as_temporary() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let _request = read_request(&mut stream).await; + tokio::time::sleep(Duration::from_secs(1)).await; + }); + let client = + RoutingClient::with_timeout(format!("http://{address}"), Duration::from_millis(20)) + .unwrap(); + let error = client.probe().await.unwrap_err(); + assert_eq!(error.kind(), RoutingErrorKind::Temporary); + server.abort(); + } + + #[tokio::test] + async fn failed_put_retries_exact_persisted_record() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let (url, server) = scripted_server(vec![ + ("404 Not Found", Vec::new()), + ("500 Internal Server Error", Vec::new()), + ("200 OK", Vec::new()), + ]) + .await; + let routing = RoutingClient::new(url).unwrap(); + let mut publisher = Publisher::open(keypair, routing, directory.path()).unwrap(); + + assert!(publisher.publish(test_cid(1)).await.is_err()); + let persisted = std::fs::read(publisher.state_path()).unwrap(); + publisher.retry_pending(test_cid(1)).await.unwrap(); + + let requests = server.await.unwrap(); + assert_eq!(requests[0].method, "GET"); + assert_eq!(requests[1].method, "PUT"); + assert_eq!(requests[2].method, "PUT"); + assert_eq!(requests[1].body, persisted); + assert_eq!(requests[2].body, persisted); + } + + #[tokio::test] + async fn higher_remote_same_value_reconciles_without_regressing_sequence() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let local = Record::new( + &keypair, + format!("/ipfs/{}", test_cid(1)), + Utc::now() + chrono::Duration::hours(2), + 3, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + RecordStore::publisher(directory.path(), name) + .persist(&local) + .unwrap(); + let desired = test_cid(1); + let remote = Record::new( + &keypair, + format!("/ipfs/{desired}"), + Utc::now() + chrono::Duration::hours(3), + 7, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + let (url, server) = scripted_server(vec![("200 OK", remote), ("200 OK", Vec::new())]).await; + let mut restarted = + Publisher::open(keypair, RoutingClient::new(url).unwrap(), directory.path()).unwrap(); + restarted.publish(desired).await.unwrap(); + + let requests = server.await.unwrap(); + let published = SignedRecord::decode(name, requests[1].body.clone()).unwrap(); + assert_eq!(published.sequence(), 7); + assert_eq!(published.deployment_cid(), Some(desired)); + assert_eq!( + std::fs::read(restarted.state_path()).unwrap(), + requests[1].body + ); + } + + #[tokio::test] + async fn publisher_fails_on_same_sequence_different_value() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let local = Record::new( + &keypair, + format!("/ipfs/{}", test_cid(1)), + Utc::now() + chrono::Duration::hours(2), + 5, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + RecordStore::publisher(directory.path(), name) + .persist(&local) + .unwrap(); + let remote = Record::new( + &keypair, + format!("/ipfs/{}", test_cid(2)), + Utc::now() + chrono::Duration::hours(3), + 5, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + let (url, server) = + scripted_server(vec![("200 OK", remote.clone()), ("200 OK", remote)]).await; + let mut publisher = + Publisher::open(keypair, RoutingClient::new(url).unwrap(), directory.path()).unwrap(); + + let error = publisher.publish(test_cid(3)).await.unwrap_err(); + assert!(format!("{error:#}").contains("equivocation")); + assert!(!publisher.has_pending_put()); + let retry_error = publisher.publish(test_cid(3)).await.unwrap_err(); + assert!(format!("{retry_error:#}").contains("equivocation")); + assert_eq!(std::fs::read(publisher.state_path()).unwrap(), local); + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 2, "equivocation retries must only GET"); + assert!(requests.iter().all(|request| request.method == "GET")); + } + + #[tokio::test] + async fn same_value_refresh_preserves_sequence_and_extends_eol() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let original = Record::new( + &keypair, + format!("/ipfs/{}", test_cid(1)), + Utc::now() + chrono::Duration::hours(1), + 11, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + let original_record = SignedRecord::decode(name, original.clone()).unwrap(); + RecordStore::publisher(directory.path(), name) + .persist(&original) + .unwrap(); + let (url, server) = + scripted_server(vec![("200 OK", original), ("200 OK", Vec::new())]).await; + let mut publisher = + Publisher::open(keypair, RoutingClient::new(url).unwrap(), directory.path()).unwrap(); + publisher.refresh(test_cid(1)).await.unwrap(); + + let requests = server.await.unwrap(); + let refreshed = SignedRecord::decode(name, requests[1].body.clone()).unwrap(); + assert_eq!(refreshed.sequence(), 11); + assert_eq!(refreshed.deployment_cid(), Some(test_cid(1))); + assert!(refreshed.eol() > original_record.eol()); + } + + #[tokio::test] + async fn higher_remote_different_value_is_adopted_and_blocks_publication() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let desired = test_cid(1); + let local = Record::new( + &keypair, + format!("/ipfs/{desired}"), + Utc::now() + chrono::Duration::hours(2), + 3, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + RecordStore::publisher(directory.path(), name) + .persist(&local) + .unwrap(); + let remote = Record::new( + &keypair, + format!("/ipfs/{}", test_cid(2)), + Utc::now() + chrono::Duration::hours(3), + 7, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + let remote_floor = remote.clone(); + let (url, server) = scripted_server(vec![("200 OK", remote)]).await; + let mut publisher = + Publisher::open(keypair, RoutingClient::new(url).unwrap(), directory.path()).unwrap(); + + let error = publisher.refresh(desired).await.unwrap_err(); + + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 1, "single-writer conflict must not PUT"); + assert_eq!(requests[0].method, "GET"); + assert!(!is_temporary_routing_failure(&error)); + assert!(format!("{error:#}").contains("single-writer conflict")); + assert!(format!("{error:#}").contains("no record was published")); + assert_eq!(publisher.current().unwrap().sequence(), 7); + assert_eq!( + publisher.current().unwrap().deployment_cid(), + Some(test_cid(2)) + ); + assert_eq!(std::fs::read(publisher.state_path()).unwrap(), remote_floor); + assert!(!publisher.has_pending_put()); + } + + #[tokio::test] + async fn refresh_does_not_regress_a_later_eol_floor() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let desired = test_cid(1); + let future = Record::new( + &keypair, + format!("/ipfs/{desired}"), + Utc::now() + chrono::Duration::hours(72), + 11, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + RecordStore::publisher(directory.path(), name) + .persist(&future) + .unwrap(); + let (url, server) = + scripted_server(vec![("200 OK", future.clone()), ("200 OK", Vec::new())]).await; + let mut publisher = + Publisher::open(keypair, RoutingClient::new(url).unwrap(), directory.path()).unwrap(); + + publisher.refresh(desired).await.unwrap(); + + let requests = server.await.unwrap(); + assert_eq!(requests[1].body, future); + assert_eq!(std::fs::read(publisher.state_path()).unwrap(), future); + } + + #[tokio::test] + async fn startup_refreshes_a_matching_record_before_initial_delay() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let desired = test_cid(1); + let near_expiry = Record::new( + &keypair, + format!("/ipfs/{desired}"), + Utc::now() + chrono::Duration::seconds(30), + 4, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap(); + RecordStore::publisher(directory.path(), name) + .persist(&near_expiry) + .unwrap(); + let (url, server) = scripted_server(vec![ + ("200 OK", near_expiry.clone()), + ("200 OK", Vec::new()), + ]) + .await; + let mut publisher = + Publisher::open(keypair, RoutingClient::new(url).unwrap(), directory.path()).unwrap(); + + publisher.publish(desired).await.unwrap(); + + let requests = server.await.unwrap(); + let refreshed = SignedRecord::decode(name, requests[1].body.clone()).unwrap(); + assert_eq!(refreshed.sequence(), 4); + assert_eq!(refreshed.deployment_cid(), Some(desired)); + assert!(refreshed.eol() > SignedRecord::decode(name, near_expiry).unwrap().eol()); + } +} diff --git a/src/lib.rs b/src/lib.rs index 2175bf00..774a4c0d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -21,6 +21,8 @@ pub mod dispatcher; #[cfg(not(target_arch = "wasm32"))] pub mod host; #[cfg(not(target_arch = "wasm32"))] +pub mod ipns; +#[cfg(not(target_arch = "wasm32"))] pub mod kernel; #[cfg(not(target_arch = "wasm32"))] pub use ipfs; diff --git a/src/stem/ipns.rs b/src/stem/ipns.rs new file mode 100644 index 00000000..248258b3 --- /dev/null +++ b/src/stem/ipns.rs @@ -0,0 +1,646 @@ +//! IPNS-backed authoritative deployment source. +//! +//! Transport uncertainty never changes authority. Signed EOL is different: +//! `next()` owns that deadline and emits one [`Update::InvalidHead`] when the +//! current binding expires without a valid replacement. + +use std::cmp::Ordering; +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::{Context, Result}; +use async_trait::async_trait; +use chrono::Utc; +use libp2p::identity::PeerId; + +use crate::ipns::{Fetch, RecordStore, RoutingClient, SignedRecord}; + +use super::{Head, InvalidHead, Source as SourceContract, Update}; + +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(5 * 60); +const DEFAULT_RETRY_BASE: Duration = Duration::from_secs(1); +const DEFAULT_RETRY_MAX: Duration = Duration::from_secs(60); + +#[derive(Clone, Debug)] +pub struct Config { + pub name: PeerId, + pub routing_url: String, + pub state_dir: PathBuf, + pub poll_interval: Duration, + pub retry_base: Duration, + pub retry_max: Duration, +} + +impl Config { + pub fn new(name: PeerId, routing_url: String, state_dir: PathBuf) -> Self { + Self { + name, + routing_url, + state_dir, + poll_interval: DEFAULT_POLL_INTERVAL, + retry_base: DEFAULT_RETRY_BASE, + retry_max: DEFAULT_RETRY_MAX, + } + } + + #[cfg(test)] + fn with_timing(mut self, poll_interval: Duration, retry: Duration) -> Self { + self.poll_interval = poll_interval; + self.retry_base = retry; + self.retry_max = retry; + self + } +} + +pub struct Source { + config: Config, + routing: RoutingClient, + store: RecordStore, + floor: Option, + last_update: Option, + authoritative: bool, + routing_probed: bool, + next_fetch: tokio::time::Instant, + retry_delay: Duration, +} + +impl Source { + pub fn new(config: Config) -> Result { + let routing = RoutingClient::new(config.routing_url.clone())?; + let store = RecordStore::follower(&config.state_dir, config.name); + let floor = store + .load()? + .map(|raw| SignedRecord::decode(config.name, raw)) + .transpose() + .context("validate persisted IPNS follower watermark")?; + let retry_delay = config.retry_base; + Ok(Self { + config, + routing, + store, + floor, + last_update: None, + authoritative: false, + routing_probed: false, + next_fetch: tokio::time::Instant::now(), + retry_delay, + }) + } + + pub fn watermark_path(&self) -> &std::path::Path { + self.store.path() + } + + fn record_update(record: &SignedRecord) -> Update { + match record.deployment_cid() { + Some(cid) => Update::Head(Head { cid }), + None => Update::InvalidHead(InvalidHead { + selected: record.value().to_vec(), + reason: + "IPNS selected a signed value that is not one /ipfs/ deployment binding" + .to_string(), + }), + } + } + + fn accept_initial_floor(&mut self) -> Option { + let floor = self.floor.as_ref()?; + if floor.is_expired_at(Utc::now()) { + tracing::info!( + ipns_sequence = floor.sequence(), + eol = %floor.eol(), + "Persisted IPNS ordering floor is expired; its value will not seed deployment" + ); + return None; + } + let update = Self::record_update(floor); + self.last_update = Some(update.clone()); + self.authoritative = true; + self.schedule_poll(); + Some(update) + } + + fn schedule_poll(&mut self) { + let ttl = self + .floor + .as_ref() + .map(SignedRecord::ttl) + .unwrap_or(self.config.poll_interval); + let delay = ttl + .min(self.config.poll_interval) + .max(Duration::from_millis(1)); + self.next_fetch = tokio::time::Instant::now() + delay; + self.retry_delay = self.config.retry_base; + } + + fn schedule_retry(&mut self) { + self.next_fetch = tokio::time::Instant::now() + self.retry_delay; + self.retry_delay = self + .retry_delay + .saturating_mul(2) + .min(self.config.retry_max); + } + + fn expiry_delay(&self) -> Option { + if !self.authoritative { + return None; + } + let eol = self.floor.as_ref()?.eol(); + Some((eol - Utc::now()).to_std().unwrap_or(Duration::ZERO)) + } + + fn expire(&mut self) -> Option { + if !self.authoritative { + return None; + } + self.authoritative = false; + let floor = self + .floor + .as_ref() + .expect("authoritative state has a floor"); + if matches!(self.last_update, Some(Update::InvalidHead(_))) { + return None; + } + let update = Update::InvalidHead(InvalidHead { + selected: floor.value().to_vec(), + reason: format!("signed IPNS binding expired at {}", floor.eol()), + }); + self.last_update = Some(update.clone()); + Some(update) + } + + /// Validate and apply one fetched raw record. + /// + /// `Some(update)` is returned only for a meaningful authoritative state + /// change. Persistence completes before memory changes or update delivery. + fn observe(&mut self, raw: Vec) -> Result> { + let candidate = SignedRecord::decode(self.config.name, raw) + .context("reject malformed or cryptographically invalid IPNS observation")?; + + if let Some(floor) = self.floor.as_ref() { + if candidate.sequence() == floor.sequence() && candidate.value() != floor.value() { + tracing::error!( + ipns_sequence = candidate.sequence(), + accepted_value = %String::from_utf8_lossy(floor.value()), + conflicting_value = %String::from_utf8_lossy(candidate.value()), + "IPNS publisher equivocation; retaining accepted binding" + ); + return Ok(None); + } + } + + if candidate.is_expired_at(Utc::now()) { + tracing::warn!( + ipns_sequence = candidate.sequence(), + eol = %candidate.eol(), + "Ignoring already-expired network IPNS record" + ); + return Ok(None); + } + + if let Some(floor) = self.floor.as_ref() { + if candidate.raw() == floor.raw() { + return Ok(None); + } + match candidate.compare(floor)? { + Ordering::Less => { + tracing::debug!( + observed_sequence = candidate.sequence(), + floor_sequence = floor.sequence(), + "Ignoring stale IPNS record" + ); + return Ok(None); + } + Ordering::Equal => return Ok(None), + Ordering::Greater => {} + } + } + + let update = Self::record_update(&candidate); + let emit = !self.authoritative || self.last_update.as_ref() != Some(&update); + + // There is no cancellation point from durable persistence through the + // state update and return to the Source caller. + self.store.persist(candidate.raw())?; + self.floor = Some(candidate); + self.authoritative = true; + self.last_update = Some(update.clone()); + self.schedule_poll(); + + Ok(emit.then_some(update)) + } + + async fn fetch_once(&mut self) -> Result> { + if !self.routing_probed { + self.routing + .probe() + .await + .context("probe Kubo HTTP Routing V1")?; + self.routing_probed = true; + } + match self.routing.get(self.config.name).await? { + Fetch::Record(raw) => self.observe(raw), + Fetch::NotFound => Ok(None), + } + } +} + +#[async_trait] +impl SourceContract for Source { + async fn current(&mut self) -> Result { + if let Some(update) = self.accept_initial_floor() { + return Ok(update); + } + match self.fetch_once().await? { + Some(update) => Ok(update), + None => anyhow::bail!( + "no current valid IPNS deployment binding exists above the durable ordering floor" + ), + } + } + + async fn next(&mut self) -> Result { + loop { + if let Some(delay) = self.expiry_delay() { + enum Event { + Expired, + Fetched(Result>), + } + let next_fetch = self.next_fetch; + let event = tokio::select! { + biased; + _ = tokio::time::sleep(delay) => Event::Expired, + result = async { + tokio::time::sleep_until(next_fetch).await; + self.fetch_once().await + } => Event::Fetched(result), + }; + match event { + Event::Expired => { + if let Some(update) = self.expire() { + return Ok(update); + } + } + Event::Fetched(Ok(Some(update))) => return Ok(update), + Event::Fetched(Ok(None)) => self.schedule_poll(), + Event::Fetched(Err(error)) => { + tracing::warn!("IPNS source observation failed; current authority remains unchanged: {error:#}"); + self.schedule_retry(); + } + } + } else { + tokio::time::sleep_until(self.next_fetch).await; + match self.fetch_once().await { + Ok(Some(update)) => return Ok(update), + Ok(None) => self.schedule_poll(), + Err(error) => { + tracing::warn!("IPNS source observation failed while no binding is authoritative: {error:#}"); + self.schedule_retry(); + } + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ipns::{RoutingClient, RECORD_TTL}; + use chrono::Duration as ChronoDuration; + use cid::multihash::Multihash; + use cid::Cid; + use libp2p::identity::Keypair; + use rust_ipns::Record; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + fn test_cid(byte: u8) -> Cid { + Cid::new_v1(0x55, Multihash::<64>::wrap(0x00, &[byte]).unwrap()) + } + + fn signed(keypair: &Keypair, cid: Cid, sequence: u64, lifetime: ChronoDuration) -> Vec { + Record::new( + keypair, + format!("/ipfs/{cid}"), + Utc::now() + lifetime, + sequence, + RECORD_TTL, + ) + .unwrap() + .encode() + .unwrap() + } + + fn source(keypair: &Keypair, directory: &std::path::Path) -> Source { + let name = keypair.public().to_peer_id(); + Source::new(Config::new( + name, + "http://127.0.0.1:1".into(), + directory.to_path_buf(), + )) + .unwrap() + } + + #[test] + fn ordering_policy_and_persist_before_emit() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let mut source = source(&keypair, directory.path()); + let first = signed(&keypair, test_cid(1), 3, ChronoDuration::hours(1)); + let first_update = source.observe(first.clone()).unwrap().unwrap(); + assert!(matches!(first_update, Update::Head(_))); + assert_eq!(std::fs::read(source.watermark_path()).unwrap(), first); + + let stale = signed(&keypair, test_cid(2), 2, ChronoDuration::hours(2)); + assert!(source.observe(stale).unwrap().is_none()); + + let duplicate = source.floor.as_ref().unwrap().raw().to_vec(); + assert!(source.observe(duplicate).unwrap().is_none()); + + let same_value = signed(&keypair, test_cid(1), 4, ChronoDuration::hours(2)); + assert!(source.observe(same_value.clone()).unwrap().is_none()); + assert_eq!(std::fs::read(source.watermark_path()).unwrap(), same_value); + + let changed = signed(&keypair, test_cid(2), 5, ChronoDuration::hours(2)); + assert!(matches!( + source.observe(changed.clone()).unwrap(), + Some(Update::Head(_)) + )); + assert_eq!(std::fs::read(source.watermark_path()).unwrap(), changed); + } + + #[test] + fn persistence_failure_cannot_advance_or_emit() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let mut source = source(&keypair, directory.path()); + std::fs::write(directory.path().join("ipns"), b"not-a-directory").unwrap(); + let candidate = signed(&keypair, test_cid(1), 1, ChronoDuration::hours(1)); + + assert!(source.observe(candidate).is_err()); + assert!(source.floor.is_none()); + assert!(source.last_update.is_none()); + assert!(!source.authoritative); + } + + #[test] + fn same_sequence_different_value_is_equivocation() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let mut source = source(&keypair, directory.path()); + let accepted = signed(&keypair, test_cid(1), 7, ChronoDuration::hours(1)); + source.observe(accepted.clone()).unwrap(); + + let conflicting = signed(&keypair, test_cid(2), 7, ChronoDuration::hours(2)); + assert!(source.observe(conflicting).unwrap().is_none()); + assert_eq!(std::fs::read(source.watermark_path()).unwrap(), accepted); + } + + #[test] + fn wrong_signer_is_rejected_without_changing_authority() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let attacker = Keypair::generate_ed25519(); + let mut source = source(&keypair, directory.path()); + let accepted = signed(&keypair, test_cid(1), 3, ChronoDuration::hours(1)); + source.observe(accepted.clone()).unwrap(); + + let forged = signed(&attacker, test_cid(2), 4, ChronoDuration::hours(2)); + assert!(source.observe(forged).is_err()); + assert_eq!(std::fs::read(source.watermark_path()).unwrap(), accepted); + assert!(matches!(source.last_update, Some(Update::Head(_)))); + } + + #[test] + fn same_value_refresh_updates_deadline_without_emitting() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let mut source = source(&keypair, directory.path()); + let original = signed(&keypair, test_cid(1), 4, ChronoDuration::minutes(30)); + source.observe(original).unwrap(); + let original_eol = source.floor.as_ref().unwrap().eol(); + + let refresh = signed(&keypair, test_cid(1), 4, ChronoDuration::hours(2)); + assert!(source.observe(refresh.clone()).unwrap().is_none()); + assert!(source.floor.as_ref().unwrap().eol() > original_eol); + assert_eq!(std::fs::read(source.watermark_path()).unwrap(), refresh); + } + + #[test] + fn later_valid_record_recovers_after_expiry() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let mut source = source(&keypair, directory.path()); + let original = signed(&keypair, test_cid(1), 1, ChronoDuration::hours(1)); + source.observe(original).unwrap(); + assert!(matches!(source.expire(), Some(Update::InvalidHead(_)))); + assert!(source.expire().is_none(), "expiry must emit only once"); + + let recovery = signed(&keypair, test_cid(1), 2, ChronoDuration::hours(2)); + assert!(matches!( + source.observe(recovery).unwrap(), + Some(Update::Head(_)) + )); + } + + #[test] + fn restart_rejects_downgrade_and_expired_value_does_not_seed() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let mut original = source(&keypair, directory.path()); + let current = signed(&keypair, test_cid(1), 9, ChronoDuration::hours(1)); + original.observe(current).unwrap(); + + let mut restarted = source(&keypair, directory.path()); + let stale = signed(&keypair, test_cid(2), 8, ChronoDuration::hours(2)); + assert!(restarted.observe(stale).unwrap().is_none()); + assert_eq!(restarted.floor.as_ref().unwrap().sequence(), 9); + + let expired_dir = tempfile::tempdir().unwrap(); + let expired = source(&keypair, expired_dir.path()); + let raw = signed(&keypair, test_cid(1), 10, ChronoDuration::seconds(-1)); + expired.store.persist(&raw).unwrap(); + let mut expired = source(&keypair, expired_dir.path()); + assert!(expired.accept_initial_floor().is_none()); + assert_eq!(expired.floor.as_ref().unwrap().sequence(), 10); + let lower = signed(&keypair, test_cid(2), 9, ChronoDuration::hours(1)); + assert!(expired.observe(lower).unwrap().is_none()); + } + + async fn response(stream: &mut tokio::net::TcpStream, status: &str, body: &[u8]) { + let header = format!( + "HTTP/1.1 {status}\r\nContent-Type: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + crate::ipns::IPNS_RECORD_MEDIA_TYPE, + body.len() + ); + stream.write_all(header.as_bytes()).await.unwrap(); + stream.write_all(body).await.unwrap(); + } + + #[tokio::test(start_paused = true)] + async fn signed_eol_wins_over_stalled_transport() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut probe, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 2048]; + let _ = probe.read(&mut request).await.unwrap(); + response(&mut probe, "400 Bad Request", b"").await; + let (mut stalled, _) = listener.accept().await.unwrap(); + let _ = stalled.read(&mut request).await.unwrap(); + std::future::pending::<()>().await; + }); + + let config = Config::new(name, format!("http://{address}"), directory.path().into()) + .with_timing(Duration::from_millis(1), Duration::from_secs(60)); + let mut source = Source::new(config).unwrap(); + let raw = signed(&keypair, test_cid(1), 1, ChronoDuration::seconds(5)); + source.observe(raw).unwrap(); + + let next = tokio::spawn(async move { source.next().await.unwrap() }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(1)).await; + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(6)).await; + let update = next.await.unwrap(); + assert!(matches!(update, Update::InvalidHead(_))); + server.abort(); + } + + #[tokio::test(start_paused = true)] + async fn transport_outage_before_eol_does_not_revoke_early() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + for status in ["400 Bad Request", "503 Service Unavailable"] { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 2048]; + let _ = stream.read(&mut request).await.unwrap(); + response(&mut stream, status, b"").await; + } + std::future::pending::<()>().await; + }); + let config = Config::new(name, format!("http://{address}"), directory.path().into()) + .with_timing(Duration::from_millis(1), Duration::from_secs(60)); + let mut source = Source::new(config).unwrap(); + source + .observe(signed( + &keypair, + test_cid(1), + 1, + ChronoDuration::seconds(10), + )) + .unwrap(); + + let next = tokio::spawn(async move { source.next().await.unwrap() }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + assert!(!next.is_finished()); + next.abort(); + server.abort(); + } + + #[tokio::test] + async fn newer_refresh_replaces_the_active_eol_deadline() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let refreshed = signed(&keypair, test_cid(1), 1, ChronoDuration::seconds(2)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (refreshed_tx, refreshed_rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + for (status, body) in [("400 Bad Request", Vec::new()), ("200 OK", refreshed)] { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 2048]; + let _ = stream.read(&mut request).await.unwrap(); + response(&mut stream, status, &body).await; + } + refreshed_tx.send(()).unwrap(); + let (_stalled, _) = listener.accept().await.unwrap(); + std::future::pending::<()>().await; + }); + let config = Config::new(name, format!("http://{address}"), directory.path().into()) + .with_timing(Duration::from_millis(1), Duration::from_secs(60)); + let mut source = Source::new(config).unwrap(); + source + .observe(signed( + &keypair, + test_cid(1), + 1, + ChronoDuration::milliseconds(500), + )) + .unwrap(); + + let next = tokio::spawn(async move { source.next().await.unwrap() }); + refreshed_rx.await.unwrap(); + tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(700)).await; + assert!(!next.is_finished(), "the replaced EOL fired"); + next.abort(); + server.abort(); + } + + #[tokio::test] + async fn next_fetch_loop_recovers_after_expiry() { + let directory = tempfile::tempdir().unwrap(); + let keypair = Keypair::generate_ed25519(); + let name = keypair.public().to_peer_id(); + let recovery = signed(&keypair, test_cid(2), 2, ChronoDuration::hours(1)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + for (status, body) in [("400 Bad Request", Vec::new()), ("200 OK", recovery)] { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 2048]; + let _ = stream.read(&mut request).await.unwrap(); + response(&mut stream, status, &body).await; + } + }); + let config = Config::new(name, format!("http://{address}"), directory.path().into()) + .with_timing(Duration::from_millis(100), Duration::from_millis(10)); + let mut source = Source::new(config).unwrap(); + source + .observe(signed( + &keypair, + test_cid(1), + 1, + ChronoDuration::milliseconds(50), + )) + .unwrap(); + + assert!(matches!( + source.next().await.unwrap(), + Update::InvalidHead(_) + )); + let recovery = tokio::time::timeout(Duration::from_secs(1), source.next()) + .await + .expect("IPNS recovery timed out") + .unwrap(); + assert!(matches!(recovery, Update::Head(_))); + server.await.unwrap(); + } + + #[tokio::test] + async fn routing_probe_distinguishes_disabled_endpoint() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 2048]; + let _ = stream.read(&mut request).await.unwrap(); + response(&mut stream, "404 Not Found", b"not found").await; + }); + let routing = RoutingClient::new(format!("http://{address}")).unwrap(); + let error = routing.probe().await.unwrap_err(); + assert_eq!(error.kind(), crate::ipns::RoutingErrorKind::Unsupported); + server.await.unwrap(); + } +} diff --git a/src/stem/mod.rs b/src/stem/mod.rs index d64775ce..cd5d5fa9 100644 --- a/src/stem/mod.rs +++ b/src/stem/mod.rs @@ -9,6 +9,7 @@ use async_trait::async_trait; use cid::Cid; pub mod atom; +pub mod ipns; /// One deployable authoritative Stem head. #[derive(Clone, Debug, Eq, PartialEq)] diff --git a/tests/cli_daemon_integration.rs b/tests/cli_daemon_integration.rs index aaab92d5..ebeb8baa 100644 --- a/tests/cli_daemon_integration.rs +++ b/tests/cli_daemon_integration.rs @@ -67,7 +67,8 @@ fn image_arguments(arguments: &[String]) -> Vec<&str> { while index < arguments.len() { match arguments[index].as_str() { - "--listen" | "--identity" | "--http-listen" | "--namespace-root" => { + "--listen" | "--identity" | "--http-listen" | "--namespace-root" + | "--ipns-routing-url" => { index += 2; } argument if argument.starts_with('-') => index += 1, @@ -300,11 +301,18 @@ async fn default_daemon_import_excludes_private_host_state() { std::fs::create_dir_all(fhs_dir.join("svc")).expect("publishable FHS tree"); std::fs::write(ww_dir.join("identity"), b"PRIVATE_HOST_IDENTITY_SENTINEL") .expect("identity fixture"); + std::fs::create_dir_all(ww_dir.join("ipns/follow")).expect("follower state directory"); + std::fs::create_dir_all(ww_dir.join("ipns/publish")).expect("publisher state directory"); std::fs::write( - ww_dir.join("durable.ipns-record"), - b"PRIVATE_MUTABLE_STATE_SENTINEL", + ww_dir.join("ipns/follow/name.record"), + b"PRIVATE_FOLLOWER_WATERMARK_SENTINEL", ) - .expect("private state fixture"); + .expect("follower watermark fixture"); + std::fs::write( + ww_dir.join("ipns/publish/name.record"), + b"PRIVATE_PUBLISHER_RECORD_SENTINEL", + ) + .expect("publisher record fixture"); std::fs::write( fhs_dir.join("svc/deployable.txt"), b"INTENDED_DEPLOYABLE_CONTENT", @@ -374,11 +382,19 @@ async fn default_daemon_import_excludes_private_host_state() { "{request}" ); assert!( - !request.contains("filename=\"durable.ipns-record\""), + !request.contains("filename=\"ipns/follow/name.record\""), + "{request}" + ); + assert!( + !request.contains("PRIVATE_FOLLOWER_WATERMARK_SENTINEL"), + "{request}" + ); + assert!( + !request.contains("filename=\"ipns/publish/name.record\""), "{request}" ); assert!( - !request.contains("PRIVATE_MUTABLE_STATE_SENTINEL"), + !request.contains("PRIVATE_PUBLISHER_RECORD_SENTINEL"), "{request}" ); } diff --git a/tests/ipns_kubo_interop.rs b/tests/ipns_kubo_interop.rs new file mode 100644 index 00000000..5b6021e7 --- /dev/null +++ b/tests/ipns_kubo_interop.rs @@ -0,0 +1,268 @@ +//! Rust/IPNS interoperability against an isolated Kubo 0.33 repository. +//! +//! Set `WW_TEST_REQUIRE_KUBO=1` for the merge-gate mode. That mode fails when +//! the exact Kubo binary is unavailable. Other local test runs skip cleanly. + +use std::path::Path; +use std::process::{Child, Command, Output, Stdio}; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use cid::multihash::Multihash; +use cid::Cid; +use libp2p::identity::Keypair; +use rust_ipns::Record; + +const REQUIRED_KUBO: &str = "0.33.0"; + +struct Kubo { + _child: KuboChild, + repository: tempfile::TempDir, + api: String, + routing_url: String, +} + +struct KuboChild(Child); + +impl Drop for KuboChild { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +impl Kubo { + fn command(repository: &Path) -> Command { + let mut command = Command::new("ipfs"); + command.env("IPFS_PATH", repository); + command + } + + fn output(repository: &Path, args: &[&str]) -> Output { + Self::command(repository) + .args(args) + .output() + .unwrap_or_else(|error| panic!("run ipfs {args:?}: {error}")) + } + + fn require_success(output: Output, operation: &str) -> String { + assert!( + output.status.success(), + "{operation} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout) + .unwrap_or_else(|error| panic!("{operation} output was not UTF-8: {error}")) + .trim() + .to_string() + } + + fn start() -> Self { + let version = Command::new("ipfs") + .args(["version", "--number"]) + .output() + .expect("WW_TEST_REQUIRE_KUBO=1 requires ipfs on PATH"); + let version = Self::require_success(version, "ipfs version"); + assert_eq!( + version, REQUIRED_KUBO, + "interoperability Kubo version drift" + ); + + let repository = tempfile::tempdir().expect("create isolated Kubo repository"); + Self::require_success( + Self::output(repository.path(), &["init", "--profile=test"]), + "isolated ipfs init", + ); + Self::require_success( + Self::output( + repository.path(), + &["config", "--json", "Gateway.ExposeRoutingAPI", "true"], + ), + "enable Kubo Routing V1", + ); + for (key, value) in [ + ("Addresses.API", "/ip4/127.0.0.1/tcp/0"), + ("Addresses.Gateway", "/ip4/127.0.0.1/tcp/0"), + ] { + Self::require_success( + Self::output(repository.path(), &["config", key, value]), + &format!("assign an ephemeral Kubo {key} listener"), + ); + } + Self::require_success( + Self::output( + repository.path(), + &[ + "config", + "--json", + "Addresses.Swarm", + r#"["/ip4/127.0.0.1/tcp/0"]"#, + ], + ), + "assign an ephemeral Kubo Swarm listener", + ); + + let child = KuboChild( + Self::command(repository.path()) + .args(["daemon", "--offline"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start isolated Kubo daemon"), + ); + let deadline = Instant::now() + Duration::from_secs(30); + let api_path = repository.path().join("api"); + let gateway_path = repository.path().join("gateway"); + while Instant::now() < deadline { + if api_path.exists() && gateway_path.exists() { + let api = std::fs::read_to_string(&api_path).expect("read isolated Kubo API file"); + let probe = Self::command(repository.path()) + .arg(format!("--api={}", api.trim())) + .arg("id") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .expect("probe isolated Kubo"); + if probe.success() { + let gateway = std::fs::read_to_string(&gateway_path) + .expect("read isolated Kubo Gateway file"); + return Self { + _child: child, + repository, + api: api.trim().to_string(), + routing_url: multiaddr_http_url(gateway.trim()), + }; + } + } + std::thread::sleep(Duration::from_millis(100)); + } + panic!("isolated Kubo did not become ready within 30 seconds"); + } + + fn cli(&self, args: &[&str]) -> String { + let mut command = Self::command(self.repository.path()); + command.arg(format!("--api={}", self.api)); + command.args(args); + Self::require_success( + command.output().expect("run isolated Kubo command"), + &format!("ipfs {args:?}"), + ) + } +} + +fn multiaddr_http_url(address: &str) -> String { + if address.starts_with("http://") || address.starts_with("https://") { + return address.to_string(); + } + let parts: Vec<_> = address.split('/').collect(); + match parts.as_slice() { + ["", "ip4", host, "tcp", port] => format!("http://{host}:{port}"), + ["", "ip6", host, "tcp", port] => format!("http://[{host}]:{port}"), + _ => panic!("unsupported Kubo listener multiaddr: {address}"), + } +} + +fn test_cid(byte: u8) -> Cid { + Cid::new_v1(0x55, Multihash::<64>::wrap(0x00, &[byte]).unwrap()) +} + +fn required() -> bool { + std::env::var_os("WW_TEST_REQUIRE_KUBO").is_some() +} + +#[tokio::test] +async fn rust_and_kubo_boxo_raw_ipns_interoperate() { + if !required() { + eprintln!("skipping Kubo interoperability; set WW_TEST_REQUIRE_KUBO=1"); + return; + } + + let kubo = Kubo::start(); + let routing = ww::ipns::RoutingClient::new(kubo.routing_url.clone()).unwrap(); + routing.probe().await.unwrap(); + + // Rust -> Kubo: Kubo has never seen this private key. + let rust_key = Keypair::generate_ed25519(); + let rust_name = rust_key.public().to_peer_id(); + let rust_canonical_name = ww::ipns::canonical_name(rust_name); + let keystore_before = kubo.cli(&["key", "list", "-l"]); + assert!(!keystore_before.contains(&rust_canonical_name)); + let rust_eol = Utc::now() + chrono::Duration::hours(48); + let rust_record = Record::new( + &rust_key, + format!("/ipfs/{}", test_cid(1)), + rust_eol, + 17, + ww::ipns::RECORD_TTL, + ) + .unwrap(); + let rust_raw = rust_record.encode().unwrap(); + routing.put(rust_name, &rust_raw).await.unwrap(); + let fetched = match routing.get(rust_name).await.unwrap() { + ww::ipns::Fetch::Record(raw) => raw, + ww::ipns::Fetch::NotFound => panic!("Kubo lost Rust-signed IPNS record"), + }; + let verified = ww::ipns::SignedRecord::decode(rust_name, fetched.clone()).unwrap(); + assert_eq!(fetched, rust_raw); + assert_eq!(verified.sequence(), 17); + assert_eq!(verified.deployment_cid(), Some(test_cid(1))); + assert_eq!(verified.ttl(), ww::ipns::RECORD_TTL); + assert_eq!(verified.eol(), rust_eol); + let decoded = Record::decode(&fetched).unwrap(); + decoded.verify(rust_name).unwrap(); + assert!(decoded.has_signature_v1()); + assert!(decoded.has_signature_v2()); + let keystore_after = kubo.cli(&["key", "list", "-l"]); + assert_eq!(keystore_before, keystore_after); + assert!(!keystore_after.contains(&rust_canonical_name)); + + // Kubo/Boxo -> Rust: cover V2-only Ed25519, hybrid Ed25519, and a + // non-inline RSA signer. Each record is fetched as raw protobuf. + for (label, key_type, v1_compat, value) in [ + ("ed-v2", "ed25519", "false", test_cid(2)), + ("ed-hybrid", "ed25519", "true", test_cid(3)), + ("rsa-v2", "rsa", "false", test_cid(4)), + ] { + let mut key_args = vec!["key", "gen", "--type", key_type, "--ipns-base", "base36"]; + if key_type == "rsa" { + key_args.extend(["--size", "2048"]); + } + key_args.push(label); + let name = kubo.cli(&key_args); + let published_after = Utc::now() + chrono::Duration::hours(47); + let published_before = Utc::now() + chrono::Duration::hours(49); + kubo.cli(&[ + "name", + "publish", + "--allow-offline", + "--resolve=false", + "--lifetime=48h", + "--ttl=5m", + &format!("--v1compat={v1_compat}"), + &format!("--key={label}"), + &format!("/ipfs/{value}"), + ]); + + let peer_id = ww::ipns::parse_name(&name).unwrap(); + assert_eq!(ww::ipns::canonical_name(peer_id), name); + let raw = match routing.get(peer_id).await.unwrap() { + ww::ipns::Fetch::Record(raw) => raw, + ww::ipns::Fetch::NotFound => panic!("Kubo-published record was not routable"), + }; + let record = Record::decode(&raw).unwrap(); + record.verify(peer_id).unwrap(); + assert_eq!(record.value(), format!("/ipfs/{value}").as_bytes()); + assert_eq!(record.sequence(), 0); + assert_eq!(record.ttl(), ww::ipns::RECORD_TTL.as_nanos() as u64); + let eol = record.validity().unwrap(); + assert!(eol > published_after); + assert!(eol < published_before); + assert_eq!(record.has_signature_v1(), v1_compat == "true"); + assert!(record.has_signature_v2()); + let wetware = ww::ipns::SignedRecord::decode(peer_id, raw).unwrap(); + assert_eq!(wetware.deployment_cid(), Some(value)); + assert_eq!(wetware.sequence(), 0); + assert_eq!(wetware.ttl(), ww::ipns::RECORD_TTL); + assert_eq!(wetware.eol(), eol); + } +}