diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 80958fa802..61318459da 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3224,11 +3224,11 @@ async fn tokio_main() -> Result<()> { let (buzz_event, admission_author_override) = if kind_u32 == buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE { - let Some(wake) = buzz_core::workflow_wake::WorkflowMentionWake::parse( + let Some(wake) = workflow_wake::authenticate( &buzz_event.event, - ) - .ok() - else { + workflow_relay_pubkey, + ) else { + tracing::warn!("workflow wake authentication failed"); continue; }; let authority = match ctx diff --git a/crates/buzz-acp/src/workflow_wake.rs b/crates/buzz-acp/src/workflow_wake.rs index dafbfe772b..b25caaeb74 100644 --- a/crates/buzz-acp/src/workflow_wake.rs +++ b/crates/buzz-acp/src/workflow_wake.rs @@ -48,6 +48,14 @@ pub fn requires_verified_wake(event: &Event, relay_pubkey: PublicKey) -> bool { && single_tag(event, "workflow-step").is_some() } +/// Authenticate and parse a relay-signed workflow wake before any authority lookup. +pub fn authenticate(wake_event: &Event, relay_pubkey: PublicKey) -> Option { + if wake_event.pubkey != relay_pubkey || wake_event.verify().is_err() { + return None; + } + WorkflowMentionWake::parse(wake_event).ok() +} + /// Verify every authority edge and return the visible message plus its signed author principal. pub fn verify( wake_event: &Event, @@ -56,10 +64,7 @@ pub fn verify( agent_pubkey: PublicKey, subscription_channel: Uuid, ) -> Option<(Event, String)> { - if wake_event.pubkey != relay_pubkey || wake_event.verify().is_err() { - return None; - } - let wake = WorkflowMentionWake::parse(wake_event).ok()?; + let wake = authenticate(wake_event, relay_pubkey)?; if wake.recipient() != agent_pubkey || authority.workflow_owner != authority.definition.pubkey.to_hex() || wake.run_id() != authority.run_id @@ -257,6 +262,23 @@ mod tests { )); } + #[test] + fn rejects_forged_wake_before_authority_lookup() { + let fixture = Fixture::valid(); + let forged = WorkflowMentionWake::new( + fixture.agent.public_key(), + fixture.channel, + fixture.run, + fixture.definition.id, + fixture.message.id, + ) + .sign(&Keys::generate()) + .expect("forged wake"); + + assert!(authenticate(&forged, fixture.relay.public_key()).is_none()); + assert!(authenticate(&fixture.wake, fixture.relay.public_key()).is_some()); + } + #[test] fn accepts_exact_authority_and_returns_signed_owner() { let fixture = Fixture::valid(); diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index f142bb7c8b..f73dbcef07 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -153,7 +153,7 @@ pub const RESULT_GATED_KINDS: &[u32] = &[ /// `#p` values exactly equal the authenticated reader's pubkey. For stored /// (non-ephemeral) kinds in this set, the storage layer additionally writes a /// NULL `search_tsv` so the event is unsearchable through NIP-50 FTS -/// (`schema/schema.sql` and `migrations/0001_initial_schema.sql` — drift +/// (`schema/schema.sql` and the forward FTS migrations — drift /// caught by `p_gated_persistent_kinds_have_storage_null_tsvector` in /// `crates/buzz-search/tests/fts_integration.rs`). /// @@ -847,6 +847,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { | KIND_DM_VISIBILITY | KIND_THREAD_SUMMARY | KIND_WINDOW_BOUNDS + | KIND_WORKFLOW_MENTION_WAKE ) } @@ -919,8 +920,9 @@ mod tests { } #[test] - fn nip43_membership_snapshot_is_relay_only() { + fn relay_generated_kinds_are_relay_only() { assert!(is_relay_only_kind(KIND_NIP43_MEMBERSHIP_LIST)); + assert!(is_relay_only_kind(KIND_WORKFLOW_MENTION_WAKE)); assert!(!is_relay_only_kind(KIND_NIP43_LEAVE_REQUEST)); } diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 2b5334c5a5..c2f8fd6f16 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -911,7 +911,7 @@ mod postgres_tests { assert!(migrations[32].sql.as_str().contains("search_tsv")); assert!(!migrations[0].sql.as_str().contains("30179")); assert!(include_str!("../../../../schema/schema.sql") - .contains("kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200)")); + .contains("kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200, 44620)")); // Public push-gateway authority is intentionally deployment-global and // durable: immediate revocation and hostile-relay admission cannot be @@ -1194,6 +1194,13 @@ mod postgres_tests { 2 ); + // Durable workflow wakes are recipient-gated and must remain outside + // full-text search on both fresh and brownfield databases. + assert_eq!(migrations[35].version, 36); + let workflow_wake_fts = migrations[35].sql.as_str(); + assert!(workflow_wake_fts.contains("kind = 44620")); + assert!(desired_schema.contains("44200, 44620")); + // pgschema intentionally reconciles DDL, not seed DML or table storage // parameters. Its post-apply reconciliation must restore and verify // both parts of the live heartbeat contract for fresh bootstraps. diff --git a/crates/buzz-search/tests/postgres_fts_integration.rs b/crates/buzz-search/tests/postgres_fts_integration.rs index 175a01aaaa..4ff5e039dc 100644 --- a/crates/buzz-search/tests/postgres_fts_integration.rs +++ b/crates/buzz-search/tests/postgres_fts_integration.rs @@ -30,8 +30,14 @@ const MIGRATION_0008_SQL: &str = const MIGRATION_0014_SQL: &str = include_str!("../../../migrations/0014_push_lease_fts.sql"); const MIGRATION_0033_SQL: &str = include_str!("../../../migrations/0033_private_managed_agent_fts.sql"); +const MIGRATION_0036_SQL: &str = + include_str!("../../../migrations/0036_workflow_mention_wake_fts.sql"); async fn setup() -> (PgPool, String) { + setup_with_search_policy(true).await +} + +async fn setup_with_search_policy(apply_fresh_allowlist: bool) -> (PgPool, String) { let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string()); let schema = format!("fts_test_{}", Uuid::new_v4().simple()); // Connect to the default schema first to create the test schema. @@ -77,15 +83,20 @@ async fn setup() -> (PgPool, String) { pool.execute(MIGRATION_0007_SQL) .await .expect("apply 0007 migration"); - pool.execute(MIGRATION_0008_SQL) - .await - .expect("apply 0008 migration"); + if apply_fresh_allowlist { + pool.execute(MIGRATION_0008_SQL) + .await + .expect("apply 0008 migration"); + } pool.execute(MIGRATION_0014_SQL) .await .expect("apply 0014 migration"); pool.execute(MIGRATION_0033_SQL) .await .expect("apply 0033 migration"); + pool.execute(MIGRATION_0036_SQL) + .await + .expect("apply 0036 migration"); (pool, schema) } @@ -1414,8 +1425,8 @@ async fn author_only_kinds_are_storage_level_unsearchable() { /// search entry point could surface tokenized content from these kinds. The /// L1 NULL tsvector is the unbreakable backstop: `@@` mathematically cannot /// match NULL. This test catches the drift where someone adds a persistent -/// kind to `P_GATED_KINDS` without the matching `schema/schema.sql` + -/// `migrations/0001_initial_schema.sql` skip-set update. +/// kind to `P_GATED_KINDS` without the matching desired schema and forward +/// migration exclusion. /// /// Ephemeral kinds (20000–29999) are skipped: they are never stored, so the /// storage-layer defense does not apply to them regardless of the schema @@ -1428,7 +1439,10 @@ async fn author_only_kinds_are_storage_level_unsearchable() { #[tokio::test] #[ignore = "requires Postgres"] async fn p_gated_persistent_kinds_have_storage_null_tsvector() { - let (pool, schema) = setup().await; + // Exercise the brownfield negative skip-set. The fresh-install positive + // allowlist would make every unknown kind unsearchable and let a missing + // per-kind migration pass vacuously. + let (pool, schema) = setup_with_search_policy(false).await; let c = mk_community(&pool, "p-gated-tripwire.example").await; let token = "pgated_tripwire_marker_qwerty"; diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index b119d26774..d1f6f0db9a 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/crates/buzz-test-client/tests/e2e_relay.rs @@ -222,40 +222,54 @@ async fn test_connect_and_authenticate() { #[tokio::test] #[ignore] -async fn test_client_submitted_nip43_membership_snapshots_are_rejected() { +async fn test_client_submitted_relay_only_events_are_rejected() { let url = relay_url(); let keys = Keys::generate(); - // Prove this actor can submit a normal event so the rejection below is + // Prove this actor can submit a normal event so the rejections below are // specifically the relay-only invariant, not a broader authorization failure. - create_test_channel(&keys).await; - let forged = EventBuilder::new(Kind::Custom(13_534), "") - .tags([Tag::parse(["member", &keys.public_key().to_hex(), "owner"]).unwrap()]) - .sign_with_keys(&keys) - .expect("sign forged membership snapshot"); + let channel_id = create_test_channel(&keys).await; + let forged_events = [ + EventBuilder::new(Kind::Custom(13_534), "") + .tags([Tag::parse(["member", &keys.public_key().to_hex(), "owner"]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign forged membership snapshot"), + EventBuilder::new(Kind::Custom(44_620), "") + .tags([ + Tag::parse(["p", &keys.public_key().to_hex()]).unwrap(), + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["run", &Uuid::new_v4().to_string()]).unwrap(), + Tag::parse(["definition", &"11".repeat(32)]).unwrap(), + Tag::parse(["message", &"22".repeat(32)]).unwrap(), + ]) + .sign_with_keys(&keys) + .expect("sign forged workflow wake"), + ]; let mut ws = BuzzTestClient::connect(&url, &keys).await.expect("connect"); - let ok = ws - .send_event(forged.clone()) - .await - .expect("submit forged snapshot via websocket"); - assert!(!ok.accepted, "forged WebSocket snapshot must be rejected"); - assert_eq!(ok.message, "restricted: relay-only kind"); - ws.disconnect().await.expect("disconnect"); + for forged in forged_events { + let ok = ws + .send_event(forged.clone()) + .await + .expect("submit forged relay-only event via websocket"); + assert!(!ok.accepted, "forged WebSocket event must be rejected"); + assert_eq!(ok.message, "restricted: relay-only kind"); - let response = reqwest::Client::new() - .post(format!("{}/events", relay_http_url())) - .header("X-Pubkey", keys.public_key().to_hex()) - .header("Content-Type", "application/json") - .body(serde_json::to_string(&forged).unwrap()) - .send() - .await - .expect("submit forged snapshot via HTTP"); - assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); - let body = response.text().await.expect("read HTTP rejection"); - assert!( - body.contains("restricted: relay-only kind"), - "unexpected HTTP rejection: {body}" - ); + let response = reqwest::Client::new() + .post(format!("{}/events", relay_http_url())) + .header("X-Pubkey", keys.public_key().to_hex()) + .header("Content-Type", "application/json") + .body(serde_json::to_string(&forged).unwrap()) + .send() + .await + .expect("submit forged relay-only event via HTTP"); + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + let body = response.text().await.expect("read HTTP rejection"); + assert!( + body.contains("restricted: relay-only kind"), + "unexpected HTTP rejection: {body}" + ); + } + ws.disconnect().await.expect("disconnect"); } #[tokio::test] diff --git a/migrations/0041_workflow_mention_wake_fts.sql b/migrations/0041_workflow_mention_wake_fts.sql new file mode 100644 index 0000000000..a0c42c95ed --- /dev/null +++ b/migrations/0041_workflow_mention_wake_fts.sql @@ -0,0 +1,32 @@ +-- Kind:44620 is a durable, recipient-gated workflow mention wake. Its canonical +-- content is empty, but keep the storage-level full-text-search backstop aligned +-- with every persistent P_GATED_KINDS member, including on brownfield databases +-- that retain the legacy negative skip-set. +-- +-- Preserve the database's existing search policy for every other kind. As with +-- 0014 and 0033, replacing this generated column rewrites the events table and +-- rebuilds the GIN index under an ACCESS EXCLUSIVE lock. +DO $$ +DECLARE + existing_expression TEXT; +BEGIN + SELECT pg_get_expr(d.adbin, d.adrelid) + INTO existing_expression + FROM pg_attrdef d + JOIN pg_attribute a + ON a.attrelid = d.adrelid + AND a.attnum = d.adnum + WHERE d.adrelid = 'events'::regclass + AND a.attname = 'search_tsv'; + + IF existing_expression IS NULL THEN + RAISE EXCEPTION 'events.search_tsv generated expression not found'; + END IF; + + ALTER TABLE events DROP COLUMN search_tsv; + EXECUTE format( + 'ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (CASE WHEN kind = 44620 THEN NULL::tsvector ELSE (%s) END) STORED', + existing_expression + ); + CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv); +END $$; diff --git a/schema/schema.sql b/schema/schema.sql index 67d5d6730b..af5289b74e 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -219,9 +219,9 @@ CREATE TABLE events ( -- Privacy: encrypted/private routing wrappers and p-gated membership notices -- must never be discoverable through NIP-50 full-text search. NULL tsvector -- never matches `@@`. - -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033). + -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033 + 0036). search_tsv TSVECTOR GENERATED ALWAYS AS ( - CASE WHEN kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector + CASE WHEN kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200, 44620) THEN NULL::tsvector ELSE to_tsvector('simple', content) END ) STORED,