jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 04:51:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflow): block banned owners from webhook runs

Browse Source 
Signed-off-by: seiler <seiler@squareup.com>
Co-authored-by: Codex <noreply@openai.com>
(cherry picked from commit 6ecd74515c01c407f817f7c25597337fcffa688a)
Signed-off-by: Brad Seiler <seiler@squareup.com>
This commit is contained in:
seiler
2026-10-05 22:45:27 +00:00
committed by Brad Seiler
parent 19aa619fdd
commit 2eb1fc279c
2 changed files with 382 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
crates/buzz-relay/src/api/bridge.rs
View File

@@ -27,6 +27,10 @@ use super::{api_error, db_read_error, internal_error, not_found, parse_query_or_
mod thread_roots;
mod thread_window;
#[cfg(test)]
#[path = "bridge/workflow_webhook_postgres_tests.rs"]
mod workflow_webhook_postgres_tests;
pub(crate) async fn enforce_http_admission(
state: &AppState,
tenant: &TenantContext,
@@ -2394,7 +2398,11 @@ pub struct WebhookQuery {
/// Webhook trigger endpoint. No user auth — the webhook secret authenticates the caller.
///
/// Prefers `X-Webhook-Secret` header over `?secret=` query param (headers aren't logged
/// by most proxies). Returns 202 Accepted; execution is async.
/// by most proxies). Returns 202 Accepted; execution is async. A run is admitted
/// when its `workflow_runs` row is inserted; immediately before that insert,
/// admission checks the workflow owner's active ban in the host-bound community.
/// This point-in-time check does not cancel already-created runs, and a
/// concurrent ban committed after the lookup can race with row insertion.
pub async fn workflow_webhook(
State(state): State<Arc<AppState>>,
Path(id_str): Path<String>,
@@ -2510,6 +2518,21 @@ pub async fn workflow_webhook(
.await
.map_err(|_| not_found("workflow not found"))?;
// The workflow_runs insert is the admission boundary. This is the last
// authorization gate before it. The restriction read is authoritative and
// tenant-scoped; lookup failures deny admission just like an active ban. It
// is intentionally a point-in-time check rather than a transaction with ban
// writes, so a ban racing after this read may follow the run insertion. Runs
// already inserted are unaffected.
let restrictions = state
.db
.moderation_restriction_state(community_id, &workflow.owner_pubkey)
.await
.map_err(|_| not_found("workflow not found"))?;
if restrictions.banned {
return Err(not_found("workflow not found"));
}
let run_id = state
.db
.create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref())

358
crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs Normal file
View File

@@ -0,0 +1,358 @@
use std::sync::Arc;
use axum::{
body::Bytes,
extract::{Path, Query, State},
http::{header, HeaderMap, HeaderValue, StatusCode},
Json,
};
use serde_json::{json, Value};
use sqlx::PgPool;
use uuid::Uuid;
use super::{workflow_webhook, WebhookQuery};
use crate::{
config::Config,
state::{AppState, AuditShutdownHandle},
test_support::database_url,
};
async fn app_state(db: buzz_db::Db, pool: &PgPool) -> (Arc<AppState>, AuditShutdownHandle) {
let mut config = Config::for_test();
config.database_url = database_url();
config.redis_url =
std::env::var("REDIS_URL").expect("REDIS_URL must point to the isolated test Redis");
config.relay_url = "wss://workflow-webhook-test.invalid".to_owned();
config.require_auth_token = false;
config.require_relay_membership = false;
let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
.create_pool(Some(deadpool_redis::Runtime::Tokio1))
.expect("create Redis pool");
let pubsub = Arc::new(
buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
.await
.expect("create pubsub manager"),
);
let audit = buzz_audit::AuditService::new(pool.clone());
let auth = buzz_auth::AuthService::new(config.auth.clone());
let search = buzz_search::SearchService::new(pool.clone());
let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
db.clone(),
buzz_workflow::WorkflowConfig::default(),
));
let media_storage = buzz_media::MediaStorage::new(&config.media).expect("create media storage");
let (state, audit_shutdown) = AppState::new(
config,
db,
redis_pool,
audit,
pubsub,
auth,
search,
workflow_engine,
nostr::Keys::generate(),
media_storage,
);
(Arc::new(state), audit_shutdown)
}
async fn seed_community(pool: &PgPool, community_id: Uuid, host: &str) {
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community_id)
.bind(host)
.execute(pool)
.await
.expect("insert community");
}
struct WebhookWorkflowSeed<'a> {
community_id: Uuid,
workflow_id: Uuid,
channel_id: Uuid,
owner_pubkey: &'a [u8],
owner_role: &'a str,
name: &'a str,
secret: &'a str,
steps: Value,
}
async fn seed_webhook_workflow(pool: &PgPool, seed: WebhookWorkflowSeed<'_>) {
sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)")
.bind(seed.community_id)
.bind(seed.owner_pubkey)
.execute(pool)
.await
.expect("insert workflow owner");
sqlx::query(
"INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, $3, $4)",
)
.bind(seed.community_id)
.bind(seed.channel_id)
.bind(format!("workflow-{}", seed.name))
.bind(seed.owner_pubkey)
.execute(pool)
.await
.expect("insert workflow channel");
sqlx::query(
"INSERT INTO channel_members (community_id, channel_id, pubkey, role) \
VALUES ($1, $2, $3, $4::member_role)",
)
.bind(seed.community_id)
.bind(seed.channel_id)
.bind(seed.owner_pubkey)
.bind(seed.owner_role)
.execute(pool)
.await
.expect("insert workflow owner membership");
let definition = json!({
"name": seed.name,
"trigger": {"on": "webhook"},
"steps": seed.steps,
"_webhook_secret": seed.secret
});
sqlx::query(
"INSERT INTO workflows \
(community_id, id, name, owner_pubkey, channel_id, definition, definition_hash) \
VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7)",
)
.bind(seed.community_id)
.bind(seed.workflow_id)
.bind(seed.name)
.bind(seed.owner_pubkey)
.bind(seed.channel_id)
.bind(definition.to_string())
.bind(vec![0u8; 32])
.execute(pool)
.await
.expect("insert workflow");
}
async fn post_webhook(
state: &Arc<AppState>,
workflow_id: Uuid,
host: &str,
secret: &str,
) -> (StatusCode, Value) {
let mut headers = HeaderMap::new();
headers.insert(
header::HOST,
HeaderValue::from_bytes(host.as_bytes()).expect("host header"),
);
headers.insert(
"x-webhook-secret",
HeaderValue::from_bytes(secret.as_bytes()).expect("secret header"),
);
match workflow_webhook(
State(Arc::clone(state)),
Path(workflow_id.to_string()),
Query(WebhookQuery { secret: None }),
headers,
Bytes::new(),
)
.await
{
Ok((status, Json(body))) | Err((status, Json(body))) => (status, body),
}
}
async fn workflow_run_count(pool: &PgPool, community_id: Uuid, workflow_id: Uuid) -> i64 {
sqlx::query_scalar(
"SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2",
)
.bind(community_id)
.bind(workflow_id)
.fetch_one(pool)
.await
.expect("count workflow runs")
}
#[tokio::test]
#[ignore = "requires PostgreSQL and Redis"]
async fn webhook_admission_enforces_owner_ban_tenant_scope_and_lookup_failure() {
let pool = PgPool::connect(&database_url())
.await
.expect("connect to isolated PostgreSQL test database");
let db = buzz_db::Db::from_pool(pool.clone());
let community_a = Uuid::new_v4();
let community_b = Uuid::new_v4();
let host_a = format!("webhook-a-{}.test", Uuid::new_v4().simple());
let host_b = format!("webhook-b-{}.test", Uuid::new_v4().simple());
let owner_pubkey = [0x31; 32];
let actor_pubkey = [0x52; 32];
let member_pubkey = [0x73; 32];
let workflow_id = Uuid::new_v4();
let channel_id = Uuid::new_v4();
let secret = format!("webhook-secret-{}", Uuid::new_v4().simple());
seed_community(&pool, community_a, &host_a).await;
seed_community(&pool, community_b, &host_b).await;
// Deliberately reuse workflow and channel IDs across tenants. The webhook
// request host must choose which community's owner restriction applies.
let delay_step = || {
json!([{
"id": "hold",
"action": "delay",
"duration": "1s"
}])
};
seed_webhook_workflow(
&pool,
WebhookWorkflowSeed {
community_id: community_a,
workflow_id,
channel_id,
owner_pubkey: &owner_pubkey,
owner_role: "owner",
name: "owner-workflow-a",
secret: &secret,
steps: delay_step(),
},
)
.await;
seed_webhook_workflow(
&pool,
WebhookWorkflowSeed {
community_id: community_b,
workflow_id,
channel_id,
owner_pubkey: &owner_pubkey,
owner_role: "owner",
name: "owner-workflow-b",
secret: &secret,
steps: delay_step(),
},
)
.await;
let member_workflow_id = Uuid::new_v4();
let member_channel_id = Uuid::new_v4();
seed_webhook_workflow(
&pool,
WebhookWorkflowSeed {
community_id: community_a,
workflow_id: member_workflow_id,
channel_id: member_channel_id,
owner_pubkey: &member_pubkey,
owner_role: "member",
name: "member-exfiltration-workflow",
secret: &secret,
steps: json!([{
"id": "send",
"action": "call_webhook",
"url": "https://workflow-webhook-test.invalid"
}]),
},
)
.await;
let (state, audit_shutdown) = app_state(db.clone(), &pool).await;
let (status, _) = post_webhook(&state, workflow_id, &host_a, "wrong-secret").await;
assert_eq!(
status,
StatusCode::UNAUTHORIZED,
"the secret remains required"
);
assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 0);
let (status, _) = post_webhook(&state, workflow_id, &host_a, &secret).await;
assert_eq!(
status,
StatusCode::ACCEPTED,
"an unbanned owner is admitted"
);
assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 1);
db.ban_community_member(
buzz_core::CommunityId::from_uuid(community_a),
&owner_pubkey,
&actor_pubkey,
None,
None,
)
.await
.expect("ban workflow owner");
let (status, body) = post_webhook(&state, workflow_id, &host_a, &secret).await;
assert_eq!(
status,
StatusCode::NOT_FOUND,
"a banned owner is denied generically"
);
assert_eq!(body["error"], "workflow not found");
assert_eq!(
workflow_run_count(&pool, community_a, workflow_id).await,
1,
"denial must not enqueue a workflow run"
);
let (status, _) = post_webhook(&state, workflow_id, &host_b, &secret).await;
assert_eq!(
status,
StatusCode::ACCEPTED,
"a ban in community A must not block the same owner in community B"
);
assert_eq!(workflow_run_count(&pool, community_b, workflow_id).await, 1);
assert!(db
.unban_community_member(
buzz_core::CommunityId::from_uuid(community_a),
&owner_pubkey,
&actor_pubkey,
)
.await
.expect("unban workflow owner"));
let (status, _) = post_webhook(&state, workflow_id, &host_a, &secret).await;
assert_eq!(
status,
StatusCode::ACCEPTED,
"an explicitly unbanned owner is admitted"
);
assert_eq!(workflow_run_count(&pool, community_a, workflow_id).await, 2);
let (status, _) = post_webhook(&state, member_workflow_id, &host_a, &secret).await;
assert_eq!(
status,
StatusCode::NOT_FOUND,
"the existing elevated-role check still rejects member-owned exfiltration workflows"
);
assert_eq!(
workflow_run_count(&pool, community_a, member_workflow_id).await,
0,
"role denial must not enqueue a workflow run"
);
audit_shutdown
.drain(std::time::Duration::from_secs(1))
.await;
drop(state);
let (failing_db, admin_pool, schema) =
crate::test_support::restriction_lookup_failing_db().await;
let (failing_state, failing_audit_shutdown) = app_state(failing_db, &pool).await;
let (status, body) = post_webhook(&failing_state, workflow_id, &host_a, &secret).await;
assert_eq!(
status,
StatusCode::NOT_FOUND,
"restriction lookup failure denies admission"
);
assert_eq!(body["error"], "workflow not found");
assert_eq!(
workflow_run_count(&pool, community_a, workflow_id).await,
2,
"a failed restriction lookup must not enqueue a workflow run"
);
failing_audit_shutdown
.drain(std::time::Duration::from_secs(1))
.await;
drop(failing_state);
sqlx::raw_sql(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE")))
.execute(&admin_pool)
.await
.expect("drop restriction lookup failure schema");
}