From f804d3adb19cc657cd15645e6ad134c767d30b33 Mon Sep 17 00:00:00 2001 From: Worph Date: Sun, 25 Oct 2026 17:14:06 +0000 Subject: [PATCH] v1.10.0: MMUR2 signed claims (claims feature), DiscoveryRecord/EnrichPatch carry claims, meta-core writes post signed claims --- Cargo.toml | 18 +- src/claims.rs | 602 +++++++++++++++++++++++++++++++++++++++ src/enrich.rs | 38 ++- src/lib.rs | 6 + src/meta_core.rs | 123 ++++++-- src/query.rs | 4 + src/serve.rs | 2 + src/types.rs | 8 + tests/claim-vectors.json | 93 ++++++ 9 files changed, 864 insertions(+), 30 deletions(-) create mode 100644 src/claims.rs create mode 100644 tests/claim-vectors.json diff --git a/Cargo.toml b/Cargo.toml index 9b2e0de..0e54f8a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-feeder-sdk" -version = "1.9.0" +version = "1.10.0" edition = "2021" rust-version = "1.85" license = "MIT" @@ -37,7 +37,7 @@ default = ["feeder"] # FeederPlugin trait, hashing, the redb cache, the axum harness. On by default # so existing consumers build exactly what they always did. feeder = [ - "beacon", "filename", + "beacon", "filename", "claims", "dep:axum", "dep:hyper", "dep:reqwest", "dep:base64", "dep:futures", "dep:async-trait", "dep:anyhow", "dep:thiserror", "dep:sha2", "dep:bytes", "dep:redb", @@ -56,6 +56,15 @@ beacon = ["dep:libc"] transport = ["feeder", "dep:cid", "dep:tokio-util"] # Conformance checks any transport implementation must pass. testkit = ["transport"] +# MMUR2 signed claims (`src/claims.rs`): the `SignedClaim` type, its canonical +# bytes, signing (k256, compact low-S), verification and the writer key +# (`METAMESH_SIGNING_KEY` / `_FILE`). `feeder` enables it because the gateway +# wire (`DiscoveryRecord`, `EnrichPatch`) carries claims; a meta-* service that +# only needs claims takes `default-features = false, features = ["claims"]`. +claims = [ + "dep:k256", "dep:bs58", "dep:sha2", "dep:base64", "dep:rand_core", + "dep:anyhow", "dep:thiserror", +] [dependencies] # Async runtime + HTTP server (the serve_feeders harness) @@ -80,6 +89,11 @@ bytes = { version = "1", optional = true } # Title-tag parsing (shared filename_meta — season/episode/quality/codec/languages). regex = { version = "1", optional = true } +# MMUR2 claims (the `claims` feature). +k256 = { version = "0.13", features = ["ecdsa", "std"], optional = true } +bs58 = { version = "0.5", optional = true } +rand_core = { version = "0.6", features = ["getrandom"], optional = true } + # Per-plugin midhash/blob cache. Schema is identical to meta-share v1 — see # cache.rs. Pure-Rust single-process embedded store. redb = { version = "2", optional = true } diff --git a/src/claims.rs b/src/claims.rs new file mode 100644 index 0000000..b4602c6 --- /dev/null +++ b/src/claims.rs @@ -0,0 +1,602 @@ +//! MMUR2 signed claims: signer `uid` says `key = value` about content `cid`, at +//! `ts` (milliseconds, also the version). See meta-root +//! `docs/implementation/metacore-claims-model.md` §3. +//! +//! Wire spec (pinned by `tests/claim-vectors.json`, vendored from meta-core's +//! `internal/claim/testdata` — `scripts/check-claim-vectors.sh` guards it): +//! +//! ```text +//! domain = b"MMUR2\n" +//! payload = canonical CBOR array [cid, uid, key, value, ts] +//! bytes_to_sign = domain || payload +//! sig = ECDSA-secp256k1(SHA-256(bytes_to_sign)), compact r||s, low-S +//! ``` +//! +//! `value` is CBOR text (public), bytes (private envelope) or null (retraction). +//! The encoder is hand-rolled: those five shapes are all MMUR2 needs, and a +//! hand-rolled head encoder cannot drift from "smallest integer form". +//! +//! The type and its JSON form are always available with the `feeder` feature +//! (the gateway wire carries claims); signing and verification are here too, +//! behind `claims`, which `feeder` enables. + +use std::collections::BTreeMap; + +use base64::Engine as _; +use serde::{de, Deserialize, Deserializer, Serialize, Serializer}; + +/// Domain separator. Changing it invalidates every claim everywhere. +pub const DOMAIN: &[u8] = b"MMUR2\n"; + +/// A claim's value. The SQLite storage type in meta-core mirrors it. +#[derive(Clone, Debug, PartialEq, Eq, Default)] +pub enum ClaimValue { + /// Retraction (tombstone). + #[default] + Null, + /// Public plaintext. + Text(String), + /// Private envelope (XChaCha20-Poly1305), opaque to everyone but the owner. + Bytes(Vec), +} + +impl ClaimValue { + pub fn is_null(&self) -> bool { + matches!(self, ClaimValue::Null) + } + pub fn as_text(&self) -> Option<&str> { + match self { + ClaimValue::Text(s) => Some(s), + _ => None, + } + } +} + +/// One signed vote. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SignedClaim { + pub cid: String, + pub uid: String, + pub key: String, + #[serde(with = "value_json")] + pub value: ClaimValue, + pub ts: u64, + /// 64-byte compact r||s, base64 in JSON. + #[serde(with = "sig_b64")] + pub sig: Vec, +} + +impl SignedClaim { + /// The exact bytes the signature covers. + pub fn signed_bytes(&self) -> Vec { + signed_bytes(&self.cid, &self.uid, &self.key, &self.value, self.ts) + } +} + +/// `domain || canonical CBOR [cid, uid, key, value, ts]`. +pub fn signed_bytes(cid: &str, uid: &str, key: &str, value: &ClaimValue, ts: u64) -> Vec { + let extra = match value { + ClaimValue::Text(s) => s.len(), + ClaimValue::Bytes(b) => b.len(), + ClaimValue::Null => 0, + }; + let mut b = Vec::with_capacity(DOMAIN.len() + cid.len() + uid.len() + key.len() + extra + 32); + b.extend_from_slice(DOMAIN); + head(&mut b, 4, 5); + text(&mut b, cid); + text(&mut b, uid); + text(&mut b, key); + match value { + ClaimValue::Text(s) => text(&mut b, s), + ClaimValue::Bytes(bytes) => { + head(&mut b, 2, bytes.len() as u64); + b.extend_from_slice(bytes); + } + ClaimValue::Null => b.push(0xf6), + } + head(&mut b, 0, ts); + b +} + +fn text(b: &mut Vec, s: &str) { + head(b, 3, s.len() as u64); + b.extend_from_slice(s.as_bytes()); +} + +/// CBOR major type + argument, smallest form. +fn head(b: &mut Vec, major: u8, n: u64) { + let m = major << 5; + if n < 24 { + b.push(m | n as u8); + } else if n <= 0xff { + b.extend_from_slice(&[m | 24, n as u8]); + } else if n <= 0xffff { + b.push(m | 25); + b.extend_from_slice(&(n as u16).to_be_bytes()); + } else if n <= 0xffff_ffff { + b.push(m | 26); + b.extend_from_slice(&(n as u32).to_be_bytes()); + } else { + b.push(m | 27); + b.extend_from_slice(&n.to_be_bytes()); + } +} + +/// JSON (human-readable) uses the spec §3.4 form — a string, `{"b64": …}` or +/// `null`. Binary formats (the bincode gateway wire) use a plain tagged enum: +/// bincode cannot decode the untagged JSON shape. +mod value_json { + use super::*; + + #[derive(Serialize, Deserialize)] + struct B64 { + b64: String, + } + + #[derive(Serialize, Deserialize)] + enum Wire { + Null, + Text(String), + Bytes(Vec), + } + + pub fn serialize(v: &ClaimValue, s: S) -> Result { + if !s.is_human_readable() { + let w = match v { + ClaimValue::Null => Wire::Null, + ClaimValue::Text(t) => Wire::Text(t.clone()), + ClaimValue::Bytes(b) => Wire::Bytes(b.clone()), + }; + return w.serialize(s); + } + match v { + ClaimValue::Null => s.serialize_none(), + ClaimValue::Text(t) => s.serialize_str(t), + ClaimValue::Bytes(b) => B64 { + b64: base64::engine::general_purpose::STANDARD.encode(b), + } + .serialize(s), + } + } + + pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result { + if !d.is_human_readable() { + return Ok(match Wire::deserialize(d)? { + Wire::Null => ClaimValue::Null, + Wire::Text(t) => ClaimValue::Text(t), + Wire::Bytes(b) => ClaimValue::Bytes(b), + }); + } + #[derive(Deserialize)] + #[serde(untagged)] + enum Raw { + Text(String), + Bytes(B64), + } + Ok(match Option::::deserialize(d)? { + None => ClaimValue::Null, + Some(Raw::Text(t)) => ClaimValue::Text(t), + Some(Raw::Bytes(B64 { b64 })) => ClaimValue::Bytes( + base64::engine::general_purpose::STANDARD + .decode(b64) + .map_err(de::Error::custom)?, + ), + }) + } +} + +/// base64 in JSON, raw bytes in binary formats. +mod sig_b64 { + use super::*; + + pub fn serialize(v: &[u8], s: S) -> Result { + if !s.is_human_readable() { + // A plain sequence, symmetric with `Vec::::deserialize` below. + return v.serialize(s); + } + s.serialize_str(&base64::engine::general_purpose::STANDARD.encode(v)) + } + + pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result, D::Error> { + if !d.is_human_readable() { + return Vec::::deserialize(d); + } + let s = String::deserialize(d)?; + base64::engine::general_purpose::STANDARD + .decode(s) + .map_err(de::Error::custom) + } +} + +/// Union-flatten a claim list into a field map: every key that has at least +/// one non-retracted text vote, with the first such value seen. A +/// **superset view for filtering and scoring**, not a merge — the merge (most +/// signers → oldest ts → smallest value, own vote first) lives only in +/// meta-core (`POST /api/merge`). Ciphertext never appears. +pub fn union_fields(claims: &[SignedClaim]) -> BTreeMap { + let mut out = BTreeMap::new(); + for c in claims { + if let ClaimValue::Text(v) = &c.value { + out.entry(c.key.clone()).or_insert_with(|| v.clone()); + } + } + out +} + +#[cfg(feature = "claims")] +pub use crypto::*; + +#[cfg(feature = "claims")] +mod crypto { + use super::*; + + use k256::ecdsa::signature::hazmat::{PrehashSigner, PrehashVerifier}; + use k256::ecdsa::{Signature, SigningKey, VerifyingKey}; + use sha2::{Digest, Sha256}; + + /// Why a claim failed verification. + #[derive(Debug, thiserror::Error)] + pub enum ClaimError { + #[error("claim: missing cid or key")] + Shape, + #[error("uid: {0}")] + Uid(String), + #[error("sig: want a 64-byte compact signature")] + SigLen, + #[error("sig: high-S signatures are refused (MMUR2 is low-S only)")] + HighS, + #[error("sig: does not verify")] + Invalid, + } + + /// Recover the verifying key a uid encodes (`"z" + base58btc(33-byte sec1)`). + pub fn pubkey_from_uid(uid: &str) -> Result { + let body = uid + .strip_prefix('z') + .ok_or_else(|| ClaimError::Uid("missing multibase 'z' prefix".into()))?; + let raw = bs58::decode(body) + .into_vec() + .map_err(|e| ClaimError::Uid(e.to_string()))?; + if raw.len() != 33 { + return Err(ClaimError::Uid(format!( + "pubkey is {} bytes, want 33", + raw.len() + ))); + } + VerifyingKey::from_sec1_bytes(&raw).map_err(|e| ClaimError::Uid(e.to_string())) + } + + /// The uid of a verifying key. + pub fn uid_from_pubkey(vk: &VerifyingKey) -> String { + let point = vk.to_encoded_point(true); + format!("z{}", bs58::encode(point.as_bytes()).into_string()) + } + + impl SignedClaim { + /// Check shape, low-S and the signature against the uid's key. + pub fn verify(&self) -> Result<(), ClaimError> { + if self.cid.is_empty() || self.key.is_empty() { + return Err(ClaimError::Shape); + } + if self.sig.len() != 64 { + return Err(ClaimError::SigLen); + } + let sig = Signature::from_slice(&self.sig).map_err(|_| ClaimError::Invalid)?; + if sig.normalize_s().is_some() { + return Err(ClaimError::HighS); + } + let vk = pubkey_from_uid(&self.uid)?; + let digest = Sha256::digest(self.signed_bytes()); + vk.verify_prehash(&digest, &sig) + .map_err(|_| ClaimError::Invalid) + } + } + + /// A writer's identity: one key per writing container (spec D2). + #[derive(Clone)] + pub struct Signer { + key: SigningKey, + uid: String, + } + + impl std::fmt::Debug for Signer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Signer") + .field("uid", &self.uid) + .finish_non_exhaustive() + } + } + + impl Signer { + /// From a 32-byte secret, hex (optionally `0x`-prefixed). + pub fn from_hex(hex_key: &str) -> anyhow::Result { + let s = hex_key.trim(); + let s = s + .strip_prefix("0x") + .or_else(|| s.strip_prefix("0X")) + .unwrap_or(s); + let raw = decode_hex(s).ok_or_else(|| anyhow::anyhow!("signing key is not hex"))?; + if raw.len() != 32 { + anyhow::bail!("signing key must be 32 bytes, got {}", raw.len()); + } + let key = SigningKey::from_slice(&raw) + .map_err(|e| anyhow::anyhow!("invalid signing key: {e}"))?; + let uid = uid_from_pubkey(key.verifying_key()); + Ok(Signer { key, uid }) + } + + /// A fresh random key. + pub fn generate() -> Self { + let key = SigningKey::random(&mut rand_core::OsRng); + let uid = uid_from_pubkey(key.verifying_key()); + Signer { key, uid } + } + + pub fn uid(&self) -> &str { + &self.uid + } + + /// The secret as lowercase hex (for writing a generated key file). + pub fn secret_hex(&self) -> String { + self.key + .to_bytes() + .iter() + .map(|b| format!("{b:02x}")) + .collect() + } + + /// Sign one claim at `ts`. RFC 6979 deterministic, normalised to low-S. + pub fn sign(&self, cid: &str, key: &str, value: ClaimValue, ts: u64) -> SignedClaim { + let bytes = signed_bytes(cid, &self.uid, key, &value, ts); + let digest = Sha256::digest(&bytes); + let sig: Signature = self + .key + .sign_prehash(&digest) + .expect("a 32-byte prehash always signs"); + let sig = sig.normalize_s().unwrap_or(sig); + SignedClaim { + cid: cid.to_string(), + uid: self.uid.clone(), + key: key.to_string(), + value, + ts, + sig: sig.to_bytes().to_vec(), + } + } + + /// Sign every field of a flat map as text claims on `cid`, all at `ts`. + pub fn sign_fields( + &self, + cid: &str, + fields: &BTreeMap, + ts: u64, + ) -> Vec { + fields + .iter() + .map(|(k, v)| self.sign(cid, k, ClaimValue::Text(v.clone()), ts)) + .collect() + } + + /// A signed retraction of `key` on `cid`. + pub fn retract(&self, cid: &str, key: &str, ts: u64) -> SignedClaim { + self.sign(cid, key, ClaimValue::Null, ts) + } + } + + /// Milliseconds since the epoch — the `ts` of a claim signed now. Strictly + /// increasing per process: `ts` is the version, so two writes to one key in + /// the same millisecond must not make the second one `stale`. + pub fn now_ms() -> u64 { + use std::sync::atomic::{AtomicU64, Ordering}; + static LAST: AtomicU64 = AtomicU64::new(0); + let wall = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let mut last = LAST.load(Ordering::Relaxed); + loop { + let next = wall.max(last + 1); + match LAST.compare_exchange_weak(last, next, Ordering::Relaxed, Ordering::Relaxed) { + Ok(_) => return next, + Err(seen) => last = seen, + } + } + } + + /// Env var holding the signer's secret as hex. + pub const SIGNING_KEY_ENV: &str = "METAMESH_SIGNING_KEY"; + /// Env var naming a key file (bind-mounted); generated on first boot. + pub const SIGNING_KEY_FILE_ENV: &str = "METAMESH_SIGNING_KEY_FILE"; + + /// The writer's key (spec §9): `METAMESH_SIGNING_KEY` (hex) wins; else + /// `METAMESH_SIGNING_KEY_FILE`, generated and written (mode 0600) when the + /// file does not exist yet. Neither set → error: a writer must never sign + /// with a silent ephemeral identity. + pub fn load_or_create_key() -> anyhow::Result { + if let Some(hex) = env_nonempty(SIGNING_KEY_ENV) { + return Signer::from_hex(&hex); + } + let Some(path) = env_nonempty(SIGNING_KEY_FILE_ENV) else { + anyhow::bail!("no signing key: set {SIGNING_KEY_ENV} or {SIGNING_KEY_FILE_ENV}"); + }; + load_or_create_key_file(std::path::Path::new(&path)) + } + + /// Read a hex key file, or generate one there (mode 0600) if absent. + pub fn load_or_create_key_file(path: &std::path::Path) -> anyhow::Result { + match std::fs::read_to_string(path) { + Ok(s) => Signer::from_hex(&s), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + let signer = Signer::generate(); + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir)?; + } + write_secret(path, &signer.secret_hex())?; + tracing::info!(target: "meta-feeder::claims", uid = %signer.uid(), + path = %path.display(), "generated a new signing key"); + Ok(signer) + } + Err(e) => Err(anyhow::anyhow!("read signing key {}: {e}", path.display())), + } + } + + #[cfg(unix)] + fn write_secret(path: &std::path::Path, hex: &str) -> std::io::Result<()> { + use std::io::Write; + use std::os::unix::fs::OpenOptionsExt; + let mut f = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(path)?; + f.write_all(hex.as_bytes())?; + f.write_all(b"\n") + } + + #[cfg(not(unix))] + fn write_secret(path: &std::path::Path, hex: &str) -> std::io::Result<()> { + std::fs::write(path, format!("{hex}\n")) + } + + fn env_nonempty(key: &str) -> Option { + std::env::var(key) + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + } + + fn decode_hex(s: &str) -> Option> { + if s.len() % 2 != 0 { + return None; + } + (0..s.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&s[i..i + 2], 16).ok()) + .collect() + } + + #[cfg(test)] + mod tests { + use super::*; + + #[derive(Deserialize)] + struct Vector { + name: String, + cid: String, + uid: String, + key: String, + value: Option, + #[serde(rename = "valueBytesHex")] + value_bytes_hex: Option, + ts: u64, + #[serde(rename = "signedBytesHex")] + signed: String, + #[serde(rename = "sigHex")] + sig: String, + } + #[derive(Deserialize)] + struct HighS { + vector: String, + #[serde(rename = "sigHex")] + sig: String, + } + #[derive(Deserialize)] + struct File { + #[serde(rename = "privateKeyHex")] + private_key_hex: String, + uid: String, + vectors: Vec, + #[serde(rename = "highS")] + high_s: HighS, + } + + fn hex(b: &[u8]) -> String { + b.iter().map(|x| format!("{x:02x}")).collect() + } + + fn load() -> File { + serde_json::from_str(include_str!("../tests/claim-vectors.json")).unwrap() + } + + fn value(v: &Vector) -> ClaimValue { + match (&v.value_bytes_hex, &v.value) { + (Some(h), _) => ClaimValue::Bytes(decode_hex(h).unwrap()), + (None, Some(t)) => ClaimValue::Text(t.clone()), + (None, None) => ClaimValue::Null, + } + } + + #[test] + fn golden_vectors() { + let f = load(); + let signer = Signer::from_hex(&f.private_key_hex).unwrap(); + assert_eq!(signer.uid(), f.uid); + for v in &f.vectors { + let c = signer.sign(&v.cid, &v.key, value(v), v.ts); + assert_eq!(c.uid, v.uid, "{}", v.name); + assert_eq!(hex(&c.signed_bytes()), v.signed, "{}: signed bytes", v.name); + assert_eq!(hex(&c.sig), v.sig, "{}: RFC 6979 signature", v.name); + c.verify().unwrap_or_else(|e| panic!("{}: {e}", v.name)); + // JSON round trip keeps the claim verifiable. + let back: SignedClaim = + serde_json::from_str(&serde_json::to_string(&c).unwrap()).unwrap(); + assert_eq!(back, c, "{}: json", v.name); + } + let v = f + .vectors + .iter() + .find(|v| v.name == f.high_s.vector) + .unwrap(); + let high = SignedClaim { + cid: v.cid.clone(), + uid: v.uid.clone(), + key: v.key.clone(), + value: value(v), + ts: v.ts, + sig: decode_hex(&f.high_s.sig).unwrap(), + }; + assert!(matches!(high.verify(), Err(ClaimError::HighS))); + } + + #[test] + fn tamper_is_refused() { + let s = Signer::generate(); + let mut c = s.sign("bafy", "title", ClaimValue::Text("A".into()), 1); + c.value = ClaimValue::Text("B".into()); + assert!(matches!(c.verify(), Err(ClaimError::Invalid))); + } + + #[test] + fn key_file_is_generated_once() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("keys/signing.key"); + let a = load_or_create_key_file(&p).unwrap(); + let b = load_or_create_key_file(&p).unwrap(); + assert_eq!(a.uid(), b.uid()); + } + + #[test] + fn json_shapes() { + let s = Signer::generate(); + let j = + serde_json::to_value(s.sign("c", "k", ClaimValue::Bytes(vec![1, 2]), 5)).unwrap(); + assert_eq!(j["value"]["b64"], "AQI="); + let j = serde_json::to_value(s.retract("c", "k", 6)).unwrap(); + assert!(j["value"].is_null()); + } + + #[test] + fn union_keeps_first_text_and_skips_retractions_and_cipher() { + let s = Signer::generate(); + let claims = vec![ + s.sign("c", "title", ClaimValue::Text("A".into()), 1), + s.sign("c", "title", ClaimValue::Text("B".into()), 2), + s.retract("c", "gone", 3), + s.sign("c", "seek", ClaimValue::Bytes(vec![9]), 4), + ]; + let f = union_fields(&claims); + assert_eq!(f.get("title").map(String::as_str), Some("A")); + assert!(!f.contains_key("gone") && !f.contains_key("seek")); + } + } +} diff --git a/src/enrich.rs b/src/enrich.rs index b55a9f8..7a3c63d 100644 --- a/src/enrich.rs +++ b/src/enrich.rs @@ -64,6 +64,10 @@ pub struct EnrichmentConfig { pub opensubtitles_password: Option, /// Wanted subtitle languages (CSV of ISO 639-1 codes; default `en`). pub opensubtitles_languages: String, + /// This feeder's claim signer (`METAMESH_SIGNING_KEY` / `_FILE`). `None` + /// when no key is configured: every meta-core write then fails loudly + /// instead of publishing unsigned data. + pub signer: Option, } fn env_nonempty(key: &str) -> Option { @@ -120,6 +124,14 @@ impl EnrichmentConfig { opensubtitles_password: env_nonempty("OPENSUBTITLES_PASSWORD"), opensubtitles_languages: env_nonempty("OPENSUBTITLES_LANGUAGES") .unwrap_or_else(|| "en".to_string()), + signer: match crate::claims::load_or_create_key() { + Ok(s) => Some(s), + Err(e) => { + warn!(target: "meta-feeder::enrich", error = %e, + "no signing key; meta-core writes will fail"); + None + } + }, } } } @@ -218,7 +230,14 @@ impl Enricher { &self.cfg.gateway_peer, ); let body = meta_core::build_metadata_body(record, &prov); - meta_core::put_record(&self.http, &self.cfg.meta_core_url, cid, &body).await + meta_core::put_record(&self.http, &self.cfg.meta_core_url, self.signer()?, cid, &body).await + } + + fn signer(&self) -> anyhow::Result<&crate::claims::Signer> { + self.cfg + .signer + .as_ref() + .ok_or_else(|| anyhow::anyhow!("no signing key (METAMESH_SIGNING_KEY / _FILE)")) } /// Run the video enrichment pipeline for one resolved CID: filename-parser @@ -240,14 +259,14 @@ impl Enricher { Ok(()) => { // Poll for the merged fields so tmdb gets the clean title. if let Some(merged) = self - .poll_until_field(cid, "originalTitle", FILENAME_PARSE_POLL_TIMEOUT) + .poll_until_field(cid, "parsedTitle", FILENAME_PARSE_POLL_TIMEOUT) .await { meta = merged; } else { debug!( target: "meta-feeder::enrich", - cid, "filename-parser produced no originalTitle within timeout; \ + cid, "filename-parser produced no parsedTitle within timeout; \ handing tmdb the base fields" ); } @@ -324,7 +343,7 @@ impl Enricher { // Poll the primary for filename-parser's clean title, then tmdb it. let merged = self - .poll_until_field(&p.cid, "originalTitle", FILENAME_PARSE_POLL_TIMEOUT) + .poll_until_field(&p.cid, "parsedTitle", FILENAME_PARSE_POLL_TIMEOUT) .await .unwrap_or_else(|| p.base.clone()); @@ -355,9 +374,14 @@ impl Enricher { } // PUT is merge/upsert — this layers the show-level anchor onto each // file's record without disturbing its per-file fields. - if let Err(e) = - meta_core::put_record(&self.http, &self.cfg.meta_core_url, &v.cid, &anchor).await - { + let res = match self.signer() { + Ok(signer) => { + meta_core::put_record(&self.http, &self.cfg.meta_core_url, signer, &v.cid, &anchor) + .await + } + Err(e) => Err(e), + }; + if let Err(e) = res { warn!(target: "meta-feeder::enrich", cid = %v.cid, error = %e, "anchor fan PUT failed"); } } diff --git a/src/lib.rs b/src/lib.rs index d3e43c8..5a8324d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -34,6 +34,8 @@ pub mod beacon; #[cfg(feature = "feeder")] pub mod budget; +#[cfg(any(feature = "feeder", feature = "claims"))] +pub mod claims; #[cfg(feature = "feeder")] pub mod cache; #[cfg(feature = "feeder")] @@ -92,5 +94,9 @@ pub use serve::{ HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse, RedeemsResponse, }; +#[cfg(any(feature = "feeder", feature = "claims"))] +pub use claims::{ClaimValue, SignedClaim}; +#[cfg(feature = "claims")] +pub use claims::{load_or_create_key, ClaimError, Signer}; #[cfg(feature = "feeder")] pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth}; diff --git a/src/meta_core.rs b/src/meta_core.rs index 5000840..02de5f9 100644 --- a/src/meta_core.rs +++ b/src/meta_core.rs @@ -12,10 +12,12 @@ //! written by the feeder is indistinguishable from one the dispatcher wrote — //! the meta-share ingester, meta-watch, and ranking all behave the same. //! -//! `PUT /api/metadata/{cid}` is merge/upsert (confirmed: meta-core -//! `SetMetadataFlat` sets only the fields in the payload, never deletes), so the -//! feeder's base write and the enrichment plugins' later `PATCH /meta/{cid}` -//! merges accumulate on the same `/file/{cid}` hash. +//! Since meta-core v2.0.0 every write is a batch of **signed claims** to +//! `POST /api/claims` (spec `docs/implementation/metacore-claims-model.md`): +//! each field becomes one MMUR2 vote signed by this feeder's own key. Votes from +//! different writers never overwrite each other — meta-core merges them — so the +//! feeder's base write and the enrichment plugins' later claims accumulate on +//! the same cid. use std::collections::BTreeMap; use std::time::{SystemTime, UNIX_EPOCH}; @@ -23,6 +25,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use serde::Deserialize; use tracing::warn; +use crate::claims::{now_ms, ClaimValue, SignedClaim, Signer}; use crate::plugin::{HashKind, HashOutcome}; use crate::types::{DiscoveryRecord, Hash}; @@ -95,22 +98,76 @@ pub fn build_metadata_body( body } -/// `PUT /api/metadata/{cid}` — write (merge/upsert) the record body into -/// meta-core. `base_url` is the meta-core root (e.g. `http://…-core:9000`). -pub async fn put_record( +/// One per-claim verdict from `POST /api/claims`. +#[derive(Debug, Deserialize)] +pub struct ClaimStatus { + pub status: String, + #[serde(default)] + pub reason: Option, +} + +/// `POST /api/claims` — submit already-signed claims (e.g. relayed verbatim). +/// Fails on a non-2xx answer or when any claim is refused as `bad_signature` / +/// `bad_request`; `stale` and `unchanged` are normal outcomes. +pub async fn post_claims( http: &reqwest::Client, base_url: &str, - cid: &str, - body: &BTreeMap, -) -> anyhow::Result<()> { - let url = format!("{}/api/metadata/{}", base_url.trim_end_matches('/'), cid); - let resp = http.put(&url).json(body).send().await?; + claims: &[SignedClaim], +) -> anyhow::Result> { + if claims.is_empty() { + return Ok(Vec::new()); + } + let url = format!("{}/api/claims", base_url.trim_end_matches('/')); + let resp = http.post(&url).json(claims).send().await?; let status = resp.status(); if !status.is_success() { let text = resp.text().await.unwrap_or_default(); - anyhow::bail!("meta-core PUT {url} returned {status}: {text}"); + anyhow::bail!("meta-core POST {url} returned {status}: {text}"); } - Ok(()) + let verdicts: Vec = resp.json().await?; + if let Some((i, bad)) = verdicts + .iter() + .enumerate() + .find(|(_, v)| v.status == "bad_signature" || v.status == "bad_request") + { + anyhow::bail!( + "meta-core refused claim {} ({}): {} {}", + i, + claims.get(i).map(|c| c.key.as_str()).unwrap_or("?"), + bad.status, + bad.reason.as_deref().unwrap_or("") + ); + } + Ok(verdicts) +} + +/// Sign every field of `body` as a text claim on `cid` (all at "now") and +/// submit them. The claims-model replacement for the v1 `PUT /api/metadata`. +pub async fn put_record( + http: &reqwest::Client, + base_url: &str, + signer: &Signer, + cid: &str, + body: &BTreeMap, +) -> anyhow::Result<()> { + let claims = signer.sign_fields(cid, body, now_ms()); + post_claims(http, base_url, &claims).await.map(|_| ()) +} + +/// Sign retractions of `keys` on `cid` (this signer's own votes only). +pub async fn retract_keys( + http: &reqwest::Client, + base_url: &str, + signer: &Signer, + cid: &str, + keys: &[String], +) -> anyhow::Result<()> { + let ts = now_ms(); + let claims: Vec = keys + .iter() + .map(|k| signer.sign(cid, k, ClaimValue::Null, ts)) + .collect(); + post_claims(http, base_url, &claims).await.map(|_| ()) } /// Shape of meta-core's `GET /meta/{cid}` response (`{ hashId, metadata }`). @@ -162,27 +219,49 @@ pub struct FeederStore { http: reqwest::Client, meta_core_url: String, gateway_peer: String, + signer: Signer, } impl FeederStore { - pub fn new(http: reqwest::Client, meta_core_url: String, gateway_peer: String) -> Self { + pub fn new( + http: reqwest::Client, + meta_core_url: String, + gateway_peer: String, + signer: Signer, + ) -> Self { FeederStore { http, meta_core_url: meta_core_url.trim_end_matches('/').to_string(), gateway_peer, + signer, } } - /// Build from the feeder env. `None` when `META_CORE_URL` is unset (the - /// feeder then falls back to the legacy "return outcomes, gateway stores" - /// path). `gateway_peer` ← `META_GATEWAY_PEER_ID` / `HOSTNAME`. + /// Build from the feeder env. `None` when `META_CORE_URL` is unset, or when + /// no signing key is configured (`METAMESH_SIGNING_KEY` / + /// `METAMESH_SIGNING_KEY_FILE`) — the feeder then falls back to the legacy + /// "return outcomes, gateway stores" path. `gateway_peer` ← + /// `META_GATEWAY_PEER_ID` / `HOSTNAME`. pub fn from_env() -> Option { let meta_core_url = env_nonempty("META_CORE_URL")?; + let signer = match crate::claims::load_or_create_key() { + Ok(s) => s, + Err(e) => { + warn!(target: "meta-feeder::store", error = %e, + "META_CORE_URL set but no signing key; not self-publishing"); + return None; + } + }; let gateway_peer = env_nonempty("META_GATEWAY_PEER_ID") .or_else(|| env_nonempty("HOSTNAME")) .unwrap_or_else(|| "gateway-feeder".to_string()); let http = crate::common::build_http_client(30, "meta-feeder-store", None); - Some(Self::new(http, meta_core_url, gateway_peer)) + Some(Self::new(http, meta_core_url, gateway_peer, signer)) + } + + /// The uid this store signs as. + pub fn signer(&self) -> &Signer { + &self.signer } pub fn meta_core_url(&self) -> &str { @@ -262,7 +341,8 @@ impl FeederStore { } } - /// Metadata-only self-publish: `PUT /api/metadata/{cid}` with provenance. + /// Metadata-only self-publish: the record's fields (+ provenance) as signed + /// claims to `POST /api/claims`. async fn store_metadata_only( &self, upstream_id: &str, @@ -272,7 +352,7 @@ impl FeederStore { ) -> anyhow::Result<()> { let prov = Provenance::now(upstream_id, record_id, &self.gateway_peer); let body = build_metadata_body(record, &prov); - put_record(&self.http, &self.meta_core_url, cid, &body).await + put_record(&self.http, &self.meta_core_url, &self.signer, cid, &body).await } } @@ -305,6 +385,7 @@ mod tests { ("title".into(), "Big Buck Bunny".into()), ("cid_btih_v1_file".into(), "bafkPERFILE".into()), ]), + claims: Vec::new(), } } diff --git a/src/query.rs b/src/query.rs index 00f54b8..ccb8e0c 100644 --- a/src/query.rs +++ b/src/query.rs @@ -100,6 +100,10 @@ pub enum GatewaySearchEvent { record_id: String, set: BTreeMap, remove: Vec, + /// Signed form of the patch: `set` as text claims, `remove` as + /// retractions. Empty from a feeder (the gateway core signs). + #[serde(default)] + claims: Vec, }, /// Retract a previously-emitted `Base` record entirely. Drop { record_id: String }, diff --git a/src/serve.rs b/src/serve.rs index d03bb4e..ce84e4f 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -799,6 +799,7 @@ mod tests { upstream_id: "stub".into(), record_id: "1".into(), fields: BTreeMap::new(), + claims: Vec::new(), }; let events = vec![ GatewaySearchEvent::Base(rec), @@ -806,6 +807,7 @@ mod tests { record_id: "1".into(), set: BTreeMap::from([("title".to_string(), "X".to_string())]), remove: vec![], + claims: vec![], }, GatewaySearchEvent::Done, ]; diff --git a/src/types.rs b/src/types.rs index 9fbc7b6..1cde759 100644 --- a/src/types.rs +++ b/src/types.rs @@ -49,6 +49,13 @@ pub struct DiscoveryRecord { pub upstream_id: String, pub record_id: String, pub fields: BTreeMap, + /// The record's MMUR2 claims (spec §10). Empty from a feeder that returns + /// plain `fields` (the gateway core signs those as itself); filled by a + /// self-publishing feeder or by the gateway before the record goes on the + /// wire. `#[serde(default)]` keeps older feeders' JSON `/query` answers + /// decoding. + #[serde(default)] // never skip_serializing_if: bincode is positional + pub claims: Vec, } /// In-process error returned by `FeederPlugin` methods. @@ -102,6 +109,7 @@ mod tests { ("title".to_string(), "AlphaFold".to_string()), ("year".to_string(), "2021".to_string()), ]), + claims: Vec::new(), }; let j = serde_json::to_string(&r).expect("serialize"); let back: DiscoveryRecord = serde_json::from_str(&j).expect("deserialize"); diff --git a/tests/claim-vectors.json b/tests/claim-vectors.json new file mode 100644 index 0000000..bcb7656 --- /dev/null +++ b/tests/claim-vectors.json @@ -0,0 +1,93 @@ +{ + "_comment": "MMUR2 golden vectors. Canonical copy: meta-core internal/claim/testdata. Do not edit by hand; see claim_test.go.", + "privateKeyHex": "0101010101010101010101010101010101010101010101010101010101010101", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "vectors": [ + { + "name": "text", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "title", + "value": "The Matrix", + "ts": 1793001234567, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a657469746c656a546865204d61747269781b000001a17733a087", + "sigHex": "3ecac653f63cfc981564b5a5cc0db0459f2b4f19b810b6391d8f487216da0f3d2e560a2b61f4a81e320287cde753e1990d7949f4f518127f6505b8fb698ee8a8" + }, + { + "name": "empty-text", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "title", + "value": "", + "ts": 1, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a657469746c656001", + "sigHex": "6eeb9ace8afe3987006c166eb0ec564b71e23d6a1c6211fdd55e8b4779c12dbb4f38c443faa3255c2ce7c6592e9d024e4ea57b5fa23446d1d4df647947e37642" + }, + { + "name": "unicode", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "titles/jpn/攻殻機動隊", + "value": "true", + "ts": 1793001234567, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a781a7469746c65732f6a706e2fe694bbe6aebbe6a99fe58b95e99a8a64747275651b000001a17733a087", + "sigHex": "64ec6342c598782df278b49636a34da4f4c009423a584454a3284efad5426aa517d83873b09a4fdd3fd32a5e9a5fef0880126edd11db773045caeb6e634bc207" + }, + { + "name": "keyset-member", + "cid": "bagacbabaec7v3", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "cids/bafkreih5kapbjzq", + "value": "true", + "ts": 23, + "signedBytesHex": "4d4d5552320a856e6261676163626162616563377633782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a75636964732f6261666b72656968356b6170626a7a71647472756517", + "sigHex": "8f707847895ccb522f1507dbab23330bd2c567d6fd1882df3bfae65bef6601024832f0b181146dcfb089aae91f5beac47da970373ab14370c4ac41410f3bf761" + }, + { + "name": "bytes", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "seek", + "value": null, + "valueBytesHex": "00ff10ab", + "ts": 24, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a647365656b4400ff10ab1818", + "sigHex": "09671b27c03edd510029ec550b35e95e3900dd37a5d688ea04c3f0409979c7236bc4a1b5fa9d1b76265da85df24693a51f0ced528a6171c90a70b92db9f246e4" + }, + { + "name": "retraction", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "genres/Drama", + "value": null, + "ts": 255, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a6c67656e7265732f4472616d61f618ff", + "sigHex": "d0c29d8edac4b78005eaf895cd9d534cb080b22ba9f7e79a5abf4d9bd906dc494464a2b388653452bca16e351132a9a92370953fa047e693459e55b7965449ac" + }, + { + "name": "long-value", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "plot/eng", + "value": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "ts": 65536, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a68706c6f742f656e6779012c6161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161611a00010000", + "sigHex": "383b2e40f913eb5240e1eae2cfa392f967321d3701623c2b3df344984686d888198bce1c78847d1898b35f991ec3465c382c45e03b9a77f2ae7d1fd740391dd2" + }, + { + "name": "ts-64bit", + "cid": "bafkreih5kapbjzq", + "uid": "zvYNYVRtXSSDCi1rZtPP3ieuoh8cG5AscesGPYESa4VpJ", + "key": "year", + "value": "1999", + "ts": 1099511627776, + "signedBytesHex": "4d4d5552320a85706261666b72656968356b6170626a7a71782d7a76594e5956527458535344436931725a745050336965756f683863473541736365734750594553613456704a647965617264313939391b0000010000000000", + "sigHex": "2b9b42ef4ebd7fc8d3ca0a514679ca4104f2e627da3a84c7b9664b27ebbdf73946cc37a04e4ea15dbed23a0e48e8797c92fcba5598b012305a314bf776398e0d" + } + ], + "highS": { + "note": "same claim as `vector`, S replaced by n-S: valid ECDSA, but MMUR2 verifiers must refuse it", + "vector": "text", + "sigHex": "3ecac653f63cfc981564b5a5cc0db0459f2b4f19b810b6391d8f487216da0f3dd1a9f5d49e0b57e1cdfd783218ac1e65ad3592f1ba308dbc5acca59166a75899" + } +}