diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f894c0e12fb..7f819bc03ae 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -745,6 +745,17 @@ jobs: --run-ignored ignored-only env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Web-of-Trust event lifecycle contract + # Verifies the persistence/query semantics of kinds 1985, 10040 and + # 30382 against real Postgres: accumulation, replacement, equal-time + # ties, author/coordinate isolation, deletion, and stale-replay fences. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-db) and test(web_of_trust_event_lifecycles_persist_and_query_exactly)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Upload relay log if: failure() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index 564cd74e9dd..2f8df1d0018 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -178,8 +178,10 @@ jobs: outputs: type=image,name=${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=${{ github.event_name != 'pull_request' }} cache-from: | type=registry,ref=${{ env.IMAGE_NAME }}-buildcache:${{ matrix.arch }} + # Only block/buzz owns the default cache namespace. Fork repositories + # may still consume its public cache, but must not try to update it. cache-to: | - ${{ (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && format('type=registry,ref={0}-buildcache:{1},mode=max,compression=zstd', env.IMAGE_NAME, matrix.arch) || '' }} + ${{ github.repository == 'block/buzz' && (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && format('type=registry,ref={0}-buildcache:{1},mode=max,compression=zstd', env.IMAGE_NAME, matrix.arch) || '' }} - name: Build and push debug image by digest id: build-debug @@ -402,7 +404,7 @@ jobs: labels: ${{ steps.meta.outputs.labels }} outputs: type=image,name=ghcr.io/block/buzz-push-gateway,push-by-digest=true,name-canonical=true,push=${{ github.event_name != 'pull_request' }} cache-from: type=registry,ref=ghcr.io/block/buzz-push-gateway-buildcache:${{ matrix.arch }} - cache-to: ${{ (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && format('type=registry,ref=ghcr.io/block/buzz-push-gateway-buildcache:{0},mode=max,compression=zstd', matrix.arch) || '' }} + cache-to: ${{ github.repository == 'block/buzz' && (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && format('type=registry,ref=ghcr.io/block/buzz-push-gateway-buildcache:{0},mode=max,compression=zstd', matrix.arch) || '' }} - name: Export digest if: github.event_name != 'pull_request' env: diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 55f727f62a6..9535725aa80 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -229,17 +229,23 @@ async fn fetch_relay_dkg_capabilities(relay_url: &str) -> Result DkgCapabilities { - for attempt in 0..DKG_CAPABILITY_ATTEMPTS { + let attempts = DKG_CAPABILITY_RETRY_DELAYS + .iter() + .copied() + .map(Some) + .chain(std::iter::once(None)); + for (attempt, retry_delay) in attempts.enumerate() { match fetch_relay_dkg_capabilities(relay_url).await { Ok(capabilities) => return capabilities, - Err(error) if attempt + 1 < DKG_CAPABILITY_ATTEMPTS => { - let delay = DKG_CAPABILITY_RETRY_DELAYS[attempt]; - tracing::warn!(attempt = attempt + 1, %error, ?delay, "relay capability discovery failed; retrying"); - tokio::time::sleep(delay).await; - } - Err(error) => { - tracing::warn!(attempt = attempt + 1, %error, "relay capability discovery failed; DKG memory disabled for this agent session"); - } + Err(error) => match retry_delay { + Some(delay) => { + tracing::warn!(attempt = attempt + 1, %error, ?delay, "relay capability discovery failed; retrying"); + tokio::time::sleep(delay).await; + } + None => { + tracing::warn!(attempt = attempt + 1, %error, "relay capability discovery failed; DKG memory disabled for this agent session"); + } + }, } } DkgCapabilities::default() diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index cbb14173c98..52f85f59c3b 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4727,19 +4727,32 @@ impl Db { // Check for the newest existing event. ORDER BY + LIMIT 1 is defensive against // historical data where prior bugs may have left multiple live rows. - let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( + // A NIP-09 delete must not let a previously superseded provider list + // become live again when an old signed event is replayed. Keep the + // latest soft-deleted kind:10040 row as an ordering fence. Other + // replaceable kinds retain their historical live-head behavior. + let preserve_deleted_ordering_fence = + kind_i32 == buzz_core::kind::KIND_TRUST_PROVIDER_LIST as i32; + let existing_sql = if preserve_deleted_ordering_fence { + "SELECT created_at, id FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ + ORDER BY created_at DESC, id ASC LIMIT 1" + } else { "SELECT created_at, id FROM events \ WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ AND channel_id IS NOT DISTINCT FROM $4 \ AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(channel_id) - .fetch_optional(&mut *tx) - .await?; + ORDER BY created_at DESC, id ASC LIMIT 1" + }; + let existing: Option<(chrono::DateTime, Vec)> = + sqlx::query_as(existing_sql) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(channel_id) + .fetch_optional(&mut *tx) + .await?; // Stale-write protection: reject if incoming is not newer. // NIP-16: created_at is second-resolution. On same-second tie, lowest @@ -5089,17 +5102,28 @@ impl Db { // Check the live head and, for NIP-RS, the compact historical ordering // watermark. The watermark remains after a NIP-09 coordinate deletion, // preventing a previously accepted signed blob from being resurrected. - let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( + // The trusted-assertion coordinate keeps its most recent soft-deleted + // version as an ordering fence. Otherwise a delayed, superseded + // kind:30382 event could resurrect after NIP-09 deletion. + let preserve_deleted_ordering_fence = + kind_i32 == buzz_core::kind::KIND_USER_TRUSTED_ASSERTION as i32; + let existing_sql = if preserve_deleted_ordering_fence { + "SELECT created_at, id FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \ + ORDER BY created_at DESC, id ASC LIMIT 1" + } else { "SELECT created_at, id FROM events \ WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .fetch_optional(&mut *tx) - .await?; + ORDER BY created_at DESC, id ASC LIMIT 1" + }; + let existing: Option<(chrono::DateTime, Vec)> = + sqlx::query_as(existing_sql) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(d_tag) + .fetch_optional(&mut *tx) + .await?; let watermark: Option<(chrono::DateTime, Vec)> = if is_nip_rs { sqlx::query_as( "SELECT created_at, event_id FROM parameterized_event_watermarks \ @@ -5351,6 +5375,348 @@ mod tests { id } + fn event_at( + keys: &nostr::Keys, + kind: u32, + content: &str, + tags: Vec, + created_at: u64, + ) -> nostr::Event { + nostr::EventBuilder::new(nostr::Kind::Custom(kind as u16), content) + .tags(tags) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("sign lifecycle event") + } + + async fn query_wot_events( + db: &Db, + community: CommunityId, + kind: u32, + author: &nostr::Keys, + d_tag: Option<&str>, + ) -> Vec { + db.query_events(&crate::event::EventQuery { + kinds: Some(vec![kind as i32]), + pubkey: Some(author.public_key().to_bytes().to_vec()), + d_tag: d_tag.map(str::to_owned), + global_only: true, + limit: Some(20), + ..crate::event::EventQuery::for_community(community) + }) + .await + .expect("query Web of Trust events") + .into_iter() + .map(|stored| stored.event) + .collect() + } + + /// Persistence/query acceptance matrix for the three Web-of-Trust kinds. + /// + /// This is selected explicitly by the Postgres-backed CI job. It proves + /// accumulation for regular labels, latest-wins ordering for replaceable + /// source lists and parameterized assertions, deterministic equal-time + /// ties, coordinate/author isolation, deletion visibility, and protection + /// against stale resurrection after deletion. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn web_of_trust_event_lifecycles_persist_and_query_exactly() { + use buzz_core::kind::{KIND_LABEL, KIND_TRUST_PROVIDER_LIST, KIND_USER_TRUSTED_ASSERTION}; + use nostr::{Keys, Tag}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let base = nostr::Timestamp::now().as_secs().saturating_sub(1_000); + + // kind:1985 is regular: independent vouches accumulate, then deletion + // hides exactly the selected event without affecting its sibling. + let label_author = Keys::generate(); + let subject = Keys::generate().public_key().to_hex(); + let label_tags = vec![ + Tag::parse(["L", "buzz.wot"]).expect("namespace tag"), + Tag::parse(["l", "vouch", "buzz.wot"]).expect("label tag"), + Tag::parse(["p", subject.as_str()]).expect("subject tag"), + ]; + let label_a = event_at( + &label_author, + KIND_LABEL, + "reviewed the parser", + label_tags.clone(), + base + 1, + ); + let label_b = event_at( + &label_author, + KIND_LABEL, + "validated the release", + label_tags, + base + 2, + ); + for event in [&label_a, &label_b] { + assert!( + db.insert_event(community, event, None) + .await + .expect("store label") + .1 + ); + } + let labels = query_wot_events(&db, community, KIND_LABEL, &label_author, None).await; + assert_eq!(labels.len(), 2, "regular labels must accumulate"); + assert!(db + .soft_delete_event(community, label_a.id.as_bytes()) + .await + .expect("delete one label")); + let labels = query_wot_events(&db, community, KIND_LABEL, &label_author, None).await; + assert_eq!(labels.len(), 1); + assert_eq!(labels[0].id, label_b.id); + + // kind:10040 is replaceable per author. Older arrivals lose; equal + // timestamps use the lowest event id; a second author is isolated. + let source_author = Keys::generate(); + let source_old = event_at( + &source_author, + KIND_TRUST_PROVIDER_LIST, + "old sources", + vec![], + base + 10, + ); + let source_new = event_at( + &source_author, + KIND_TRUST_PROVIDER_LIST, + "new sources", + vec![], + base + 20, + ); + assert!( + db.replace_addressable_event(community, &source_new, None) + .await + .expect("store new source list") + .1 + ); + assert!( + !db.replace_addressable_event(community, &source_old, None) + .await + .expect("reject stale source list") + .1 + ); + let tie_a = event_at( + &source_author, + KIND_TRUST_PROVIDER_LIST, + "tie a", + vec![], + base + 30, + ); + let tie_b = event_at( + &source_author, + KIND_TRUST_PROVIDER_LIST, + "tie b", + vec![], + base + 30, + ); + let (tie_winner, tie_loser) = if tie_a.id < tie_b.id { + (&tie_a, &tie_b) + } else { + (&tie_b, &tie_a) + }; + assert!( + db.replace_addressable_event(community, tie_loser, None) + .await + .expect("store first tie") + .1 + ); + assert!( + db.replace_addressable_event(community, tie_winner, None) + .await + .expect("lower id wins tie") + .1 + ); + assert!( + !db.replace_addressable_event(community, tie_loser, None) + .await + .expect("higher id stays stale") + .1 + ); + let other_source_author = Keys::generate(); + let other_source = event_at( + &other_source_author, + KIND_TRUST_PROVIDER_LIST, + "other author's sources", + vec![], + base + 25, + ); + assert!( + db.replace_addressable_event(community, &other_source, None) + .await + .expect("store isolated author") + .1 + ); + let sources = query_wot_events( + &db, + community, + KIND_TRUST_PROVIDER_LIST, + &source_author, + None, + ) + .await; + assert_eq!( + sources.iter().map(|event| event.id).collect::>(), + vec![tie_winner.id] + ); + assert!(db + .soft_delete_event(community, tie_winner.id.as_bytes()) + .await + .expect("delete source head")); + assert!( + !db.replace_addressable_event(community, tie_loser, None) + .await + .expect("deleted head fences stale replay") + .1 + ); + assert!(query_wot_events( + &db, + community, + KIND_TRUST_PROVIDER_LIST, + &source_author, + None, + ) + .await + .is_empty()); + assert_eq!( + query_wot_events( + &db, + community, + KIND_TRUST_PROVIDER_LIST, + &other_source_author, + None, + ) + .await + .len(), + 1, + "author deletion must remain isolated" + ); + + // kind:30382 is parameterized-replaceable per (author, subject d tag). + let assertion_author = Keys::generate(); + let subject_a = Keys::generate().public_key().to_hex(); + let subject_b = Keys::generate().public_key().to_hex(); + let assertion_old = event_at( + &assertion_author, + KIND_USER_TRUSTED_ASSERTION, + "old assertion", + vec![Tag::parse(["d", subject_a.as_str()]).expect("subject a")], + base + 40, + ); + let assertion_new = event_at( + &assertion_author, + KIND_USER_TRUSTED_ASSERTION, + "new assertion", + vec![Tag::parse(["d", subject_a.as_str()]).expect("subject a")], + base + 50, + ); + assert!( + db.replace_parameterized_event(community, &assertion_new, &subject_a, None) + .await + .expect("store new assertion") + .1 + ); + assert!( + !db.replace_parameterized_event(community, &assertion_old, &subject_a, None) + .await + .expect("reject stale assertion") + .1 + ); + let assertion_b = event_at( + &assertion_author, + KIND_USER_TRUSTED_ASSERTION, + "subject b assertion", + vec![Tag::parse(["d", subject_b.as_str()]).expect("subject b")], + base + 45, + ); + assert!( + db.replace_parameterized_event(community, &assertion_b, &subject_b, None) + .await + .expect("store isolated subject") + .1 + ); + let assertion_tie_a = event_at( + &assertion_author, + KIND_USER_TRUSTED_ASSERTION, + "assertion tie a", + vec![Tag::parse(["d", subject_a.as_str()]).expect("subject a")], + base + 60, + ); + let assertion_tie_b = event_at( + &assertion_author, + KIND_USER_TRUSTED_ASSERTION, + "assertion tie b", + vec![Tag::parse(["d", subject_a.as_str()]).expect("subject a")], + base + 60, + ); + let (assertion_winner, assertion_loser) = if assertion_tie_a.id < assertion_tie_b.id { + (&assertion_tie_a, &assertion_tie_b) + } else { + (&assertion_tie_b, &assertion_tie_a) + }; + assert!( + db.replace_parameterized_event(community, assertion_loser, &subject_a, None) + .await + .expect("store assertion tie") + .1 + ); + assert!( + db.replace_parameterized_event(community, assertion_winner, &subject_a, None) + .await + .expect("lower assertion id wins") + .1 + ); + let assertions_a = query_wot_events( + &db, + community, + KIND_USER_TRUSTED_ASSERTION, + &assertion_author, + Some(&subject_a), + ) + .await; + assert_eq!( + assertions_a + .iter() + .map(|event| event.id) + .collect::>(), + vec![assertion_winner.id] + ); + assert_eq!( + query_wot_events( + &db, + community, + KIND_USER_TRUSTED_ASSERTION, + &assertion_author, + Some(&subject_b), + ) + .await + .len(), + 1, + "d-tag coordinates must remain isolated" + ); + assert!(db + .soft_delete_event(community, assertion_winner.id.as_bytes()) + .await + .expect("delete assertion head")); + assert!( + !db.replace_parameterized_event(community, assertion_loser, &subject_a, None) + .await + .expect("deleted assertion fences stale replay") + .1 + ); + assert!(query_wot_events( + &db, + community, + KIND_USER_TRUSTED_ASSERTION, + &assertion_author, + Some(&subject_a), + ) + .await + .is_empty()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn nip_rs_replacement_hard_deletes_payload_and_watermark_rejects_replay() { diff --git a/crates/buzz-relay/src/api/dkg_query.rs b/crates/buzz-relay/src/api/dkg_query.rs index 85cfc8c3df8..b989c26dd27 100644 --- a/crates/buzz-relay/src/api/dkg_query.rs +++ b/crates/buzz-relay/src/api/dkg_query.rs @@ -23,7 +23,10 @@ use buzz_core::TenantContext; use crate::state::AppState; -use super::{api_error, bridge, internal_error, not_found}; +use super::{ + api_error, bridge, internal_error, not_found, + reputation_provider::{ConfiguredReputationProvider, ReputationProvider}, +}; /// Maximum public request body accepted by `/api/dkg/query`. pub(crate) const MAX_REQUEST_BYTES: usize = 16 * 1024; @@ -412,15 +415,19 @@ pub async fn query( .await?; let forward = parse_and_sanitize_request(&body, &requester)?; + enforce_authoritative_channel_read(&state, &tenant, forward.channel_id, &requester_bytes) + .await?; + if matches!( forward.operation, Operation::TrustNetwork | Operation::ReputationSummary - ) && !config.trust_enabled - { - return Err(not_found("not found")); + ) { + let request = serde_json::to_value(&forward) + .map_err(|_| internal_error("serializing reputation-provider request"))?; + let provider = + ConfiguredReputationProvider::from_dkg_config(state.config.dkg_query.as_ref()); + return provider.attestations(&request).await.into_http_result(); } - enforce_authoritative_channel_read(&state, &tenant, forward.channel_id, &requester_bytes) - .await?; let client = HTTP_CLIENT .as_ref() diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index 1511042a640..3e3791e5faf 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -11,6 +11,7 @@ pub mod media; pub mod mesh_demo; pub mod nip05; pub mod operator; +pub mod reputation_provider; // Re-export imeta helpers used by ingest pipeline. pub use crate::handlers::imeta::{validate_imeta_tags, verify_imeta_blobs}; diff --git a/crates/buzz-relay/src/api/reputation_provider.rs b/crates/buzz-relay/src/api/reputation_provider.rs new file mode 100644 index 00000000000..82d3a9e7686 --- /dev/null +++ b/crates/buzz-relay/src/api/reputation_provider.rs @@ -0,0 +1,448 @@ +//! Backend-neutral reputation-provider boundary. +//! +//! The first configured provider is the existing OriginTrail DKG query +//! gateway. The null provider is the default and reports `disabled` instead +//! of manufacturing an empty evidence set. Provider metadata is added to the +//! existing gateway envelope without changing its `result`, so current DKG +//! clients remain compatible while reputation callers gain explicit +//! resolution semantics. + +use async_trait::async_trait; +use axum::{http::StatusCode, response::Json}; +use chrono::{DateTime, Utc}; +use serde::Serialize; +use serde_json::{Map, Value}; + +use crate::config::DkgQueryConfig; + +use super::{api_error, dkg_query}; + +type ApiResponse = (StatusCode, Json); +type ApiResult = Result; + +/// Outcome of one bounded reputation-provider resolution attempt. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ResolutionState { + /// No provider is configured. This is not an empty evidence result. + Disabled, + /// Every configured source answered within the bounded request. + Complete, + /// At least one configured source could not provide a complete page. + Partial, + /// The provider could not perform the resolution attempt. + Unavailable, +} + +/// Per-source status carried with every provider result. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SourceDiagnostic { + /// Stable identifier for the contributing source. + pub source_id: &'static str, + /// Resolution reached by this source. + pub resolution: ResolutionState, + /// Safe, human-readable reason for a non-complete result. + #[serde(skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +/// Backend-neutral result returned by [`ReputationProvider`]. +#[derive(Debug)] +pub struct ReputationBatch { + /// Aggregate resolution for this request. + pub resolution: ResolutionState, + /// Stable provider identifier. + pub provider_id: &'static str, + /// Provider contract version. + pub provider_version: &'static str, + /// Time at which the resolution was produced. + pub as_of: DateTime, + /// Per-source outcomes used to produce the aggregate resolution. + pub source_diagnostics: Vec, + response: Option, +} + +impl ReputationBatch { + fn disabled() -> Self { + Self { + resolution: ResolutionState::Disabled, + provider_id: "null", + provider_version: "1", + as_of: Utc::now(), + source_diagnostics: Vec::new(), + response: None, + } + } + + fn dkg(resolution: ResolutionState, response: ApiResponse, detail: Option) -> Self { + Self { + resolution, + provider_id: "origintrail-dkg", + provider_version: "dkg-trust@1", + as_of: Utc::now(), + source_diagnostics: vec![SourceDiagnostic { + source_id: "origintrail-dkg", + resolution, + detail, + }], + response: Some(response), + } + } + + fn metadata(&self) -> Map { + let mut metadata = Map::new(); + metadata.insert( + "resolution".to_string(), + serde_json::to_value(self.resolution).expect("resolution serializes"), + ); + metadata.insert( + "providerId".to_string(), + Value::String(self.provider_id.to_string()), + ); + metadata.insert( + "providerVersion".to_string(), + Value::String(self.provider_version.to_string()), + ); + metadata.insert("asOf".to_string(), Value::String(self.as_of.to_rfc3339())); + metadata.insert( + "sourceDiagnostics".to_string(), + serde_json::to_value(&self.source_diagnostics).expect("diagnostics serialize"), + ); + metadata + } + + /// Convert the provider result back to the relay's existing HTTP shape. + /// + /// Successful DKG responses retain `ok`, `channelId`, `cg`, `operation`, + /// and `result`. The provider fields are additive. Disabled and unavailable + /// providers use the same fields on an error response, allowing a new + /// client to distinguish configuration from outage without changing the + /// legacy route-discovery behavior. + pub fn into_http_result(mut self) -> ApiResult { + let response = self.response.take(); + let metadata = self.metadata(); + let (status, Json(mut value)) = response.unwrap_or_else(|| { + api_error( + StatusCode::NOT_FOUND, + "reputation provider is not configured", + ) + }); + let Some(object) = value.as_object_mut() else { + let (status, Json(mut error)) = api_error( + StatusCode::BAD_GATEWAY, + "reputation provider returned an invalid response envelope", + ); + error + .as_object_mut() + .expect("api errors are objects") + .extend(metadata); + return Err((status, Json(error))); + }; + object.extend(metadata); + + let response = (status, Json(value)); + if status.is_success() && self.resolution != ResolutionState::Unavailable { + Ok(response) + } else { + Err(response) + } + } +} + +/// Minimal provider seam for evidence-backed reputation resolution. +#[async_trait] +pub trait ReputationProvider: Send + Sync { + /// Resolve the bounded attestation request without changing its scope. + async fn attestations(&self, request: &Value) -> ReputationBatch; +} + +/// Default provider: reputation is disabled, not empty. +#[derive(Debug, Default)] +pub struct NullProvider; + +#[async_trait] +impl ReputationProvider for NullProvider { + async fn attestations(&self, _request: &Value) -> ReputationBatch { + ReputationBatch::disabled() + } +} + +/// Adapter for the existing bounded OriginTrail DKG trust operations. +#[derive(Debug)] +pub struct DkgProvider<'a> { + config: &'a DkgQueryConfig, +} + +impl<'a> DkgProvider<'a> { + fn new(config: &'a DkgQueryConfig) -> Self { + Self { config } + } +} + +#[async_trait] +impl ReputationProvider for DkgProvider<'_> { + async fn attestations(&self, request: &Value) -> ReputationBatch { + let client = match dkg_query::HTTP_CLIENT.as_ref() { + Ok(client) => client, + Err(_) => { + return ReputationBatch::dkg( + ResolutionState::Unavailable, + api_error( + StatusCode::BAD_GATEWAY, + "reputation provider is unavailable", + ), + Some("HTTP client initialization failed".to_string()), + ); + } + }; + let response = match client + .post(self.config.url.clone()) + .bearer_auth(&self.config.bearer_token) + .header(reqwest::header::ACCEPT, "application/json") + .timeout(self.config.timeout) + .json(request) + .send() + .await + { + Ok(response) => response, + Err(error) => { + let detail = if error.is_timeout() { + "provider deadline exceeded" + } else { + "provider transport failed" + }; + return ReputationBatch::dkg( + ResolutionState::Unavailable, + dkg_query::upstream_error(error), + Some(detail.to_string()), + ); + } + }; + + match dkg_query::bounded_json_response(response).await { + Ok((status, Json(value))) if status.is_success() => { + match validate_gateway_success(request, &value) { + Ok(resolution) => ReputationBatch::dkg(resolution, (status, Json(value)), None), + Err(detail) => ReputationBatch::dkg( + ResolutionState::Unavailable, + api_error( + StatusCode::BAD_GATEWAY, + "reputation provider returned an invalid success envelope", + ), + Some(detail.to_string()), + ), + } + } + Ok(response) => ReputationBatch::dkg( + ResolutionState::Unavailable, + response, + Some("provider rejected the bounded request".to_string()), + ), + Err(response) => ReputationBatch::dkg( + ResolutionState::Unavailable, + response, + Some("provider returned an unusable response".to_string()), + ), + } + } +} + +fn validate_gateway_success( + request: &Value, + response: &Value, +) -> Result { + let expected_channel = request + .get("channelId") + .and_then(Value::as_str) + .ok_or("provider request channel is invalid")?; + let expected_operation = request + .get("operation") + .and_then(Value::as_str) + .ok_or("provider request operation is invalid")?; + if response.get("ok").and_then(Value::as_bool) != Some(true) { + return Err("provider success envelope did not assert ok=true"); + } + if response.get("channelId").and_then(Value::as_str) != Some(expected_channel) { + return Err("provider response channel did not match the authorized request"); + } + if response.get("operation").and_then(Value::as_str) != Some(expected_operation) { + return Err("provider response operation did not match the authorized request"); + } + if response + .get("cg") + .and_then(Value::as_str) + .is_none_or(str::is_empty) + { + return Err("provider response omitted the resolved Context Graph"); + } + let result = response + .get("result") + .and_then(Value::as_object) + .ok_or("provider response result was not an object")?; + match expected_operation { + "trust_network" + if result.get("people").is_some_and(Value::is_array) + && result.get("vouches").is_some_and(Value::is_array) => {} + "reputation_summary" + if result.get("subject").is_some_and(Value::is_string) + && result.get("perspective").is_some_and(Value::is_string) + && result.get("score").is_some_and(Value::is_number) + && result.get("breakdown").is_some_and(Value::is_object) + && result.get("signals").is_some_and(Value::is_object) + && result.get("evidence").is_some_and(Value::is_array) => {} + "trust_network" | "reputation_summary" => { + return Err("provider response result did not match the requested operation"); + } + _ => return Err("unsupported operation reached the reputation provider"), + } + Ok(resolution_from_gateway(response)) +} + +/// Runtime provider selection. The null provider is the deterministic default. +#[derive(Debug)] +pub enum ConfiguredReputationProvider<'a> { + /// Deterministic default when reputation is not configured. + Null(NullProvider), + /// Existing OriginTrail DKG trust-query adapter. + Dkg(DkgProvider<'a>), +} + +impl<'a> ConfiguredReputationProvider<'a> { + /// Select the configured DKG adapter or the null provider. + pub fn from_dkg_config(config: Option<&'a DkgQueryConfig>) -> Self { + match config.filter(|config| config.trust_enabled) { + Some(config) => Self::Dkg(DkgProvider::new(config)), + None => Self::Null(NullProvider), + } + } +} + +#[async_trait] +impl ReputationProvider for ConfiguredReputationProvider<'_> { + async fn attestations(&self, request: &Value) -> ReputationBatch { + match self { + Self::Null(provider) => provider.attestations(request).await, + Self::Dkg(provider) => provider.attestations(request).await, + } + } +} + +fn resolution_from_gateway(value: &Value) -> ResolutionState { + match value + .pointer("/result/completeness") + .and_then(Value::as_str) + { + Some("partial") => ResolutionState::Partial, + Some("complete") => ResolutionState::Complete, + // Absence or an unknown future value cannot safely be promoted to a + // complete evidence result. Older adapters remain usable, but callers + // see the conservative partial state. + _ => ResolutionState::Partial, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn null_provider_is_disabled_not_empty_complete() { + let provider = ConfiguredReputationProvider::from_dkg_config(None); + let batch = provider.attestations(&serde_json::json!({})).await; + assert_eq!(batch.resolution, ResolutionState::Disabled); + assert_eq!(batch.provider_id, "null"); + assert!(batch.source_diagnostics.is_empty()); + + let error = batch + .into_http_result() + .expect_err("disabled provider is not an evidence response"); + assert_eq!(error.0, StatusCode::NOT_FOUND); + assert_eq!(error.1 .0["resolution"], "disabled"); + assert_eq!(error.1 .0["providerId"], "null"); + } + + #[test] + fn gateway_completeness_maps_to_explicit_resolution_states() { + assert_eq!( + resolution_from_gateway(&serde_json::json!({ + "result": {"completeness": "complete"} + })), + ResolutionState::Complete + ); + assert_eq!( + resolution_from_gateway(&serde_json::json!({ + "result": {"completeness": "partial"} + })), + ResolutionState::Partial + ); + assert_eq!( + resolution_from_gateway(&serde_json::json!({"result": {}})), + ResolutionState::Partial + ); + } + + #[test] + fn gateway_success_is_bound_to_the_authorized_request_and_result_shape() { + let request = serde_json::json!({ + "channelId": "channel", + "operation": "trust_network" + }); + let valid = serde_json::json!({ + "ok": true, + "channelId": "channel", + "cg": "did:dkg:context-graph:channel", + "operation": "trust_network", + "result": {"completeness": "complete", "people": [], "vouches": []} + }); + assert_eq!( + validate_gateway_success(&request, &valid), + Ok(ResolutionState::Complete) + ); + + for malformed in [ + serde_json::json!({}), + serde_json::json!({"result": {"completeness": "complete"}}), + serde_json::json!({ + "ok": true, + "channelId": "other-channel", + "cg": "did:dkg:context-graph:channel", + "operation": "trust_network", + "result": {"people": [], "vouches": []} + }), + serde_json::json!({ + "ok": true, + "channelId": "channel", + "cg": "did:dkg:context-graph:channel", + "operation": "reputation_summary", + "result": {"people": [], "vouches": []} + }), + ] { + assert!(validate_gateway_success(&request, &malformed).is_err()); + } + } + + #[test] + fn provider_metadata_is_additive_to_the_existing_gateway_envelope() { + let batch = ReputationBatch::dkg( + ResolutionState::Complete, + ( + StatusCode::OK, + Json(serde_json::json!({ + "ok": true, + "channelId": "channel", + "cg": "cg", + "operation": "trust_network", + "result": {"completeness": "complete", "people": [], "vouches": []} + })), + ), + None, + ); + let (_, Json(value)) = batch.into_http_result().expect("successful batch"); + assert_eq!(value["result"]["people"], serde_json::json!([])); + assert_eq!(value["resolution"], "complete"); + assert_eq!(value["providerId"], "origintrail-dkg"); + assert_eq!(value["providerVersion"], "dkg-trust@1"); + assert!(value["asOf"].is_string()); + } +} diff --git a/desktop/src/features/dkg-memory/api.ts b/desktop/src/features/dkg-memory/api.ts index ac6f49da900..9846ddc81e6 100644 --- a/desktop/src/features/dkg-memory/api.ts +++ b/desktop/src/features/dkg-memory/api.ts @@ -7,7 +7,11 @@ import { relayClient } from "@/shared/api/relayClient"; import { getRelayHttpUrl, signRelayEvent } from "@/shared/api/tauri"; import { fetchDkgMemoryCapabilities } from "./capabilities"; -import { postAuthenticatedDkgJson, queryDkgProvider } from "./provider"; +import { + postAuthenticatedDkgJson, + queryDkgProvider, + type ReputationProviderMetadata, +} from "./provider"; import { memoryProposalProgress, normalizeMemoryProposalResponse, @@ -140,7 +144,7 @@ export interface WorkEvidence { layer: "SWM" | "VM"; } -export interface TrustNetwork { +export interface TrustNetwork extends Partial { gate: MemoryGate; cg?: string; completeness: "complete" | "partial"; @@ -148,7 +152,7 @@ export interface TrustNetwork { vouches: TrustVouch[]; } -export interface ReputationSummary { +export interface ReputationSummary extends Partial { gate: MemoryGate; cg?: string; subject: string; diff --git a/desktop/src/features/dkg-memory/provider.test.mjs b/desktop/src/features/dkg-memory/provider.test.mjs index 79e9ec9b17d..3ebe9f4f19a 100644 --- a/desktop/src/features/dkg-memory/provider.test.mjs +++ b/desktop/src/features/dkg-memory/provider.test.mjs @@ -146,6 +146,33 @@ test("community gateway uses active relay URL and a fresh payload-bound NIP-98 e assert.notEqual(nonces[0], nonces[1]); }); +test("disabled provider metadata survives an authenticated error response", async () => { + installTauri(); + globalThis.fetch = async () => + new Response( + JSON.stringify({ + error: "reputation provider is not configured", + resolution: "disabled", + providerId: "null", + providerVersion: "1", + asOf: "2026-08-12T10:00:00Z", + sourceDiagnostics: [], + }), + { status: 404 }, + ); + + await assert.rejects( + () => fetchTrustNetwork(CHANNEL_ID), + (error) => { + assert.equal(error instanceof DkgProviderError, true); + assert.equal(error.status, 404); + assert.equal(error.reputation.resolution, "disabled"); + assert.equal(error.reputation.providerId, "null"); + return true; + }, + ); +}); + test("semantic queries are explicitly current-channel scoped and expose their cost", async () => { installTauri(); let requestBody; @@ -427,6 +454,16 @@ test("web of trust uses a fixed channel-scoped operation, never client SPARQL", channelId: body.channelId, cg: "server-cg", operation: body.operation, + resolution: "complete", + providerId: "origintrail-dkg", + providerVersion: "dkg-trust@1", + asOf: "2026-08-12T10:00:00Z", + sourceDiagnostics: [ + { + sourceId: "origintrail-dkg", + resolution: "complete", + }, + ], result: { completeness: "complete", people: [ @@ -454,6 +491,9 @@ test("web of trust uses a fixed channel-scoped operation, never client SPARQL", assert.equal(JSON.stringify(body).includes("sparql"), false); assert.equal(result.gate, "ok"); assert.equal(result.cg, "server-cg"); + assert.equal(result.resolution, "complete"); + assert.equal(result.providerId, "origintrail-dkg"); + assert.equal(result.sourceDiagnostics[0].resolution, "complete"); assert.equal(result.people[0].contributions, 3); }); diff --git a/desktop/src/features/dkg-memory/provider.ts b/desktop/src/features/dkg-memory/provider.ts index 9815d73481c..e8c93c05da1 100644 --- a/desktop/src/features/dkg-memory/provider.ts +++ b/desktop/src/features/dkg-memory/provider.ts @@ -7,6 +7,26 @@ const NIP98_KIND = 27235; export type ExplorerSource = "local" | "gateway"; +export type ReputationResolution = + | "disabled" + | "complete" + | "partial" + | "unavailable"; + +export interface ReputationSourceDiagnostic { + sourceId: string; + resolution: ReputationResolution; + detail?: string; +} + +export interface ReputationProviderMetadata { + resolution: ReputationResolution; + providerId: string; + providerVersion: string; + asOf: string; + sourceDiagnostics: ReputationSourceDiagnostic[]; +} + export type DkgQueryOperation = | "channel_memory" | "contributor_trail" @@ -57,7 +77,7 @@ type CommunityGatewayEnvelope = { cg: string; operation: DkgQueryOperation; result: unknown; -}; +} & Partial; type AuthenticatedDkgPost = { path: `/api/dkg/${string}`; @@ -69,18 +89,21 @@ export class DkgProviderError extends Error { status: number; code?: string; details?: unknown; + reputation?: ReputationProviderMetadata; constructor( message: string, status: number, code?: string, details?: unknown, + reputation?: ReputationProviderMetadata, ) { super(message); this.name = "DkgProviderError"; this.status = status; this.code = code; this.details = details; + this.reputation = reputation; } } @@ -202,6 +225,7 @@ export async function postAuthenticatedDkgJson({ response.status, error.code, error.details, + reputationProviderMetadata(payload), ); } return { result: (payload ?? {}) as Result, status: response.status }; @@ -240,6 +264,46 @@ function validateEnvelope( return value as CommunityGatewayEnvelope & { operation: Operation }; } +function reputationProviderMetadata( + value: unknown, +): ReputationProviderMetadata | undefined { + if (!isRecord(value) || value.resolution === undefined) return undefined; + const resolutions = new Set([ + "disabled", + "complete", + "partial", + "unavailable", + ]); + if ( + typeof value.resolution !== "string" || + !resolutions.has(value.resolution as ReputationResolution) || + typeof value.providerId !== "string" || + typeof value.providerVersion !== "string" || + typeof value.asOf !== "string" || + !Array.isArray(value.sourceDiagnostics) + ) { + throw protocolError("invalid reputation-provider metadata"); + } + for (const diagnostic of value.sourceDiagnostics) { + if ( + !isRecord(diagnostic) || + typeof diagnostic.sourceId !== "string" || + typeof diagnostic.resolution !== "string" || + !resolutions.has(diagnostic.resolution as ReputationResolution) || + (diagnostic.detail !== undefined && typeof diagnostic.detail !== "string") + ) { + throw protocolError("invalid reputation source diagnostic"); + } + } + return { + resolution: value.resolution as ReputationResolution, + providerId: value.providerId, + providerVersion: value.providerVersion, + asOf: value.asOf, + sourceDiagnostics: value.sourceDiagnostics as ReputationSourceDiagnostic[], + }; +} + function adaptCommunityResult( envelope: CommunityGatewayEnvelope & { operation: Operation }, ): unknown { @@ -265,7 +329,12 @@ function adaptCommunityResult( case "subgraph_graph": case "subgraph_triples": case "semantic_query": - return { ...envelope.result, gate: "ok", cg: envelope.cg }; + return { + ...envelope.result, + ...(reputationProviderMetadata(envelope) ?? {}), + gate: "ok", + cg: envelope.cg, + }; case "evidence": return { ...envelope.result, gate: "ok" }; default: