relay: enforce ban/timeout write-block on command kinds

ingest_event routed command kinds — DM open/add/hide, workflow
define/trigger, approval grant/deny — straight to
command_executor::handle_command and returned before the durable
ban/timeout write-block gate that ordinary writes pass through.
handle_command performs no restriction check of its own, so a banned or
timed-out member could still open DMs, define or trigger workflows, and
grant or deny approvals over signed POST /events. The moderation-command
and relay-admin kinds are also routed around the shared gate, but each
re-checks the ban inside its own handler; command kinds had no such
check.

Enforce the gate before routing command kinds. Command kinds are
ordinary writes, not administrative capability, so they get the full
gate including timeouts — the relay-admin timeout exemption does not
extend to them. Extract the ban/timeout decision into
restriction_block/enforce_moderation_restriction so the command-kind
path and the ordinary-write path share one implementation and cannot
drift; the restriction-state lookup fails closed on a DB error.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Signed-off-by: Max Lampert <maxwell@squareup.com>

crates/buzz-relay/src/handlers/ingest.rs

@@ -1463,6 +1463,53 @@ pub async fn ingest_event(
     result
 }
 
+/// Map a durable restriction state to the write-block error it implies, or
+/// `None` if the actor may write. Banned actors are refused `blocked:`, actors
+/// with an unexpired timeout `restricted:`; an expired timeout does not block.
+fn restriction_block(r: &buzz_db::moderation::RestrictionState) -> Option<IngestError> {
+    if r.banned {
+        return Some(IngestError::AuthFailed(
+            "blocked: you are banned from this community".to_string(),
+        ));
+    }
+    if let Some(until) = r.muted_until {
+        if until > chrono::Utc::now() {
+            return Some(IngestError::AuthFailed(format!(
+                "restricted: you are timed out until {}",
+                until.timestamp()
+            )));
+        }
+    }
+    None
+}
+
+/// Enforce the durable community ban/timeout write-block for `pubkey`.
+///
+/// The lookup fails closed: a restriction-state DB error is surfaced as
+/// `error:` (500) rather than admitting the write, so a Postgres blip can never
+/// let a banned or timed-out actor through. This is the durable backstop the
+/// live ban fan-out's best-effort delivery relies on; callers decide which
+/// kinds are exempt.
+async fn enforce_moderation_restriction(
+    state: &Arc<AppState>,
+    tenant: &TenantContext,
+    pubkey: &[u8],
+) -> Result<(), IngestError> {
+    match state
+        .db
+        .moderation_restriction_state(tenant.community(), pubkey)
+        .await
+    {
+        Ok(r) => match restriction_block(&r) {
+            Some(e) => Err(e),
+            None => Ok(()),
+        },
+        Err(e) => Err(IngestError::Internal(format!(
+            "error: internal error checking restriction state: {e}"
+        ))),
+    }
+}
+
 async fn ingest_event_inner(
     state: &Arc<AppState>,
     tracer: &Arc<dyn buzz_conformance::Tracer>,
@@ -1570,7 +1617,17 @@ async fn ingest_event_inner(
 
     // Command kinds are routed AFTER signature verification, timestamp check,
     // pubkey/auth match, and scope validation — never before.
+    //
+    // They route straight to the executor, which performs no restriction check
+    // of its own, so the durable ban/timeout write-block is enforced here
+    // before routing — a banned or timed-out member must not open DMs, define
+    // or trigger workflows, or grant/deny approvals. Unlike the moderation and
+    // relay-admin exemptions below, command kinds are ordinary writes, not
+    // administrative capability, so they get the full gate including timeouts.
+    // Scope matches the ordinary-write gate below: the authoring pubkey only,
+    // with no NIP-OA owner→agent cascade.
     if buzz_core::kind::is_command_kind(kind_u32) {
+        enforce_moderation_restriction(state, tenant, auth.pubkey().as_bytes()).await?;
         return super::command_executor::handle_command(tenant, state, event, auth).await;
     }
 
@@ -1653,34 +1710,7 @@ async fn ingest_event_inner(
     // the restriction-state cache (see should-fix), which can fold in owner
     // resolution without a per-write DB round-trip.
     if !buzz_core::kind::is_moderation_command_kind(kind_u32) && !is_relay_admin_kind(kind_u32) {
-        match state
-            .db
-            .moderation_restriction_state(tenant.community(), auth.pubkey().as_bytes())
-            .await
-        {
-            Ok(r) => {
-                if r.banned {
-                    return Err(IngestError::AuthFailed(
-                        "blocked: you are banned from this community".to_string(),
-                    ));
-                }
-                if let Some(until) = r.muted_until {
-                    if until > chrono::Utc::now() {
-                        return Err(IngestError::AuthFailed(format!(
-                            "restricted: you are timed out until {}",
-                            until.timestamp()
-                        )));
-                    }
-                }
-            }
-            Err(e) => {
-                // Fail closed: a DB error must not let a banned/timed-out actor
-                // write.
-                return Err(IngestError::Internal(format!(
-                    "error: internal error checking restriction state: {e}"
-                )));
-            }
-        }
+        enforce_moderation_restriction(state, tenant, auth.pubkey().as_bytes()).await?;
     }
 
     let mut channel_id = if kind_u32 == KIND_REACTION {
@@ -2617,6 +2647,61 @@ mod tests {
         }
     }
 
+    /// A banned actor is refused with the durable `blocked:` write-block —
+    /// the error every restricted write path (including command kinds, which
+    /// route to `handle_command` and have no ban check of their own) shares.
+    #[test]
+    fn restriction_block_banned_is_blocked() {
+        let banned = buzz_db::moderation::RestrictionState {
+            banned: true,
+            muted_until: None,
+        };
+        match restriction_block(&banned) {
+            Some(IngestError::AuthFailed(msg)) => {
+                assert_eq!(msg, "blocked: you are banned from this community");
+            }
+            other => panic!("banned actor must be blocked, got {other:?}"),
+        }
+    }
+
+    /// An unexpired timeout blocks writes with `restricted:` — command kinds
+    /// are ordinary writes, so a timeout blocks them just like any post.
+    #[test]
+    fn restriction_block_active_timeout_is_restricted() {
+        let timed_out = buzz_db::moderation::RestrictionState {
+            banned: false,
+            muted_until: Some(chrono::Utc::now() + chrono::Duration::minutes(5)),
+        };
+        match restriction_block(&timed_out) {
+            Some(IngestError::AuthFailed(msg)) => {
+                assert!(
+                    msg.starts_with("restricted: you are timed out until"),
+                    "unexpected timeout message: {msg:?}"
+                );
+            }
+            other => panic!("active timeout must be restricted, got {other:?}"),
+        }
+    }
+
+    /// An expired timeout is not a write-block — the row lingers but the
+    /// `muted_until` guard clears once it is in the past.
+    #[test]
+    fn restriction_block_expired_timeout_does_not_block() {
+        let expired = buzz_db::moderation::RestrictionState {
+            banned: false,
+            muted_until: Some(chrono::Utc::now() - chrono::Duration::minutes(5)),
+        };
+        assert!(
+            restriction_block(&expired).is_none(),
+            "an expired timeout must not block writes"
+        );
+    }
+
+    #[test]
+    fn restriction_block_unrestricted_is_none() {
+        assert!(restriction_block(&buzz_db::moderation::RestrictionState::default()).is_none());
+    }
+
     #[derive(Debug, Default)]
     struct VecTracer {
         steps: Mutex<Vec<TraceStep>>,

crates/buzz-test-client/tests/regression_command_kind_ban_gate.rs

@@ -0,0 +1,269 @@
+//! Regression test for the command-kind durable-ban/timeout bypass.
+//!
+//! `ingest_event` routes command kinds (DM open/add/hide, workflow def/trigger,
+//! approval grant/deny) straight to `command_executor::handle_command` and
+//! returns *before* the durable ban/timeout write-block gate. Unlike the
+//! moderation-command and relay-admin exemptions — which each re-check the ban
+//! inside their own handler — `handle_command` performs no restriction check,
+//! so a banned or timed-out member could still open DMs, define/trigger
+//! workflows, and grant/deny approvals over signed NIP-98 `POST /events`.
+//!
+//! The gate now covers command kinds; this test pins the contract:
+//!   - a **banned** member's command kind is refused 403 + `blocked:`,
+//!   - a **timed-out** member's command kind is refused 403 + `restricted:`
+//!     (command kinds are ordinary writes, not admin capability, so a timeout
+//!     blocks them — the relay-admin exemption does NOT extend here),
+//!   - an **unrestricted** member's command kind still succeeds and mutates.
+//!
+//! Requires a running relay and its Postgres. Ignored by default:
+//!   REPRO_RELAY_HTTP=http://localhost:3030 REPRO_HOST=localhost:3030 \
+//!   DATABASE_URL=postgres://buzz:buzz_dev@localhost:5471/buzz \
+//!   cargo test -p buzz-test-client --test regression_command_kind_ban_gate \
+//!     -- --ignored --nocapture
+
+use base64::engine::general_purpose::STANDARD as BASE64;
+use base64::Engine;
+use nostr::{EventBuilder, Keys, Kind, Tag};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+const KIND_DM_OPEN: u16 = 41010;
+
+fn http_base() -> String {
+    std::env::var("REPRO_RELAY_HTTP").unwrap_or_else(|_| "http://localhost:3030".into())
+}
+fn host() -> String {
+    std::env::var("REPRO_HOST").unwrap_or_else(|_| "localhost:3030".into())
+}
+fn db_url() -> String {
+    std::env::var("DATABASE_URL").expect("DATABASE_URL required")
+}
+
+fn sha256_hex(b: &[u8]) -> String {
+    hex::encode(Sha256::digest(b))
+}
+
+fn nip98(keys: &Keys, url: &str, body: &str) -> String {
+    let ev = EventBuilder::new(Kind::Custom(27_235), "")
+        .tags(vec![
+            Tag::parse(["u", url]).unwrap(),
+            Tag::parse(["method", "POST"]).unwrap(),
+            Tag::parse(["payload", &sha256_hex(body.as_bytes())]).unwrap(),
+            Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap(),
+        ])
+        .sign_with_keys(keys)
+        .unwrap();
+    format!(
+        "Nostr {}",
+        BASE64.encode(serde_json::to_string(&ev).unwrap())
+    )
+}
+
+async fn post_event(keys: &Keys, event: &nostr::Event) -> (u16, String) {
+    let body = serde_json::to_string(event).unwrap();
+    let signed_url = format!("http://{}/events", host());
+    let r = reqwest::Client::new()
+        .post(format!("{}/events", http_base()))
+        .header("Host", host())
+        .header("Content-Type", "application/json")
+        .header("Authorization", nip98(keys, &signed_url, &body))
+        .body(body)
+        .send()
+        .await
+        .expect("POST /events");
+    let status = r.status().as_u16();
+    (status, r.text().await.unwrap_or_default())
+}
+
+fn signed(keys: &Keys, kind: u16, tags: Vec<Tag>) -> nostr::Event {
+    EventBuilder::new(Kind::Custom(kind), "")
+        .tags(tags)
+        .sign_with_keys(keys)
+        .unwrap()
+}
+
+/// A DM-open command from `actor` to `peer` (a single `p` tag is the minimum
+/// `handle_dm_open` accepts).
+fn dm_open(actor: &Keys, peer: &Keys) -> nostr::Event {
+    signed(
+        actor,
+        KIND_DM_OPEN,
+        vec![Tag::parse(["p", &peer.public_key().to_hex()]).unwrap()],
+    )
+}
+
+async fn pool() -> sqlx::Pool<sqlx::Postgres> {
+    sqlx::postgres::PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&db_url())
+        .await
+        .expect("connect Postgres")
+}
+
+async fn community_id(p: &sqlx::Pool<sqlx::Postgres>) -> Uuid {
+    let id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO communities (id, host) VALUES ($1, $2) ON CONFLICT (lower(host)) DO NOTHING",
+    )
+    .bind(id)
+    .bind(host())
+    .execute(p)
+    .await
+    .unwrap();
+    sqlx::query_scalar("SELECT id FROM communities WHERE lower(host) = lower($1)")
+        .bind(host())
+        .fetch_one(p)
+        .await
+        .unwrap()
+}
+
+async fn seed(p: &sqlx::Pool<sqlx::Postgres>, cid: Uuid, keys: &Keys, role: &str) {
+    sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2) ON CONFLICT DO NOTHING")
+        .bind(cid)
+        .bind(keys.public_key().to_bytes().to_vec())
+        .execute(p)
+        .await
+        .ok();
+    sqlx::query(
+        "INSERT INTO relay_members (community_id, pubkey, role, added_by) VALUES ($1,$2,$3,NULL) \
+         ON CONFLICT (community_id, pubkey) DO UPDATE SET role = $3, updated_at = now()",
+    )
+    .bind(cid)
+    .bind(keys.public_key().to_hex())
+    .bind(role)
+    .execute(p)
+    .await
+    .unwrap();
+}
+
+/// Count active DM (channel) memberships a pubkey holds — the observable
+/// mutation `handle_dm_open` produces. Zero means no DM was created for them.
+async fn channel_membership_count(p: &sqlx::Pool<sqlx::Postgres>, cid: Uuid, keys: &Keys) -> i64 {
+    sqlx::query_scalar::<_, i64>(
+        "SELECT count(*) FROM channel_members WHERE community_id = $1 AND pubkey = $2",
+    )
+    .bind(cid)
+    .bind(keys.public_key().to_bytes().to_vec())
+    .fetch_one(p)
+    .await
+    .unwrap()
+}
+
+/// Post-fix regression bar for command kinds.
+///
+/// Pre-fix this FAILS at the first assertion: the banned member's `DM_OPEN` is
+/// accepted (200) instead of refused (403) — the reproduction. Post-fix all
+/// three actors behave per the durable write-block contract.
+#[tokio::test]
+#[ignore]
+async fn banned_or_timed_out_member_cannot_run_a_command_kind() {
+    let p = pool().await;
+    let cid = community_id(&p).await;
+
+    let owner = Keys::generate();
+    let banned = Keys::generate();
+    let timed_out = Keys::generate();
+    let good = Keys::generate();
+    for (k, r) in [
+        (&owner, "owner"),
+        (&banned, "member"),
+        (&timed_out, "member"),
+        (&good, "member"),
+    ] {
+        seed(&p, cid, k, r).await;
+    }
+
+    // Owner bans one member and times out another via the real 9040/9042 path.
+    let (s, b) = post_event(
+        &owner,
+        &signed(
+            &owner,
+            9040,
+            vec![Tag::parse(["p", &banned.public_key().to_hex()]).unwrap()],
+        ),
+    )
+    .await;
+    assert_eq!(s, 200, "ban must land: {b}");
+    let expiry = (chrono_now() + 3600).to_string();
+    let (s, b) = post_event(
+        &owner,
+        &signed(
+            &owner,
+            9042,
+            vec![
+                Tag::parse(["p", &timed_out.public_key().to_hex()]).unwrap(),
+                Tag::parse(["expiration", &expiry]).unwrap(),
+            ],
+        ),
+    )
+    .await;
+    assert_eq!(s, 200, "timeout must land: {b}");
+
+    // Snapshot DM memberships each actor holds AFTER moderation — a ban/timeout
+    // itself sends the target a notice DM — so the mutation check below measures
+    // only what each actor's own DM_OPEN attempt adds.
+    let before_banned = channel_membership_count(&p, cid, &banned).await;
+    let before_timed_out = channel_membership_count(&p, cid, &timed_out).await;
+    let before_good = channel_membership_count(&p, cid, &good).await;
+
+    // ── Banned member: DM_OPEN must be 403 + `blocked:`. ──
+    let (bs, bb) = post_event(&banned, &dm_open(&banned, &owner)).await;
+    println!("[banned] DM_OPEN -> {bs} {bb}");
+    assert_eq!(
+        bs, 403,
+        "banned member's command kind must get 403, got {bs} {bb}"
+    );
+    let msg: serde_json::Value = serde_json::from_str(&bb).unwrap_or_default();
+    let text = msg.get("error").and_then(|v| v.as_str()).unwrap_or(&bb);
+    assert_eq!(
+        text, "blocked: you are banned from this community",
+        "banned member must carry the exact `blocked:` wire contract"
+    );
+
+    // ── Timed-out member: DM_OPEN must be 403 + `restricted:`. Command kinds are
+    // ordinary writes, not admin capability, so a timeout blocks them. ──
+    let (ts, tb) = post_event(&timed_out, &dm_open(&timed_out, &owner)).await;
+    println!("[timed-out] DM_OPEN -> {ts} {tb}");
+    assert_eq!(
+        ts, 403,
+        "timed-out member's command kind must get 403, got {ts} {tb}"
+    );
+    let tmsg: serde_json::Value = serde_json::from_str(&tb).unwrap_or_default();
+    let ttext = tmsg.get("error").and_then(|v| v.as_str()).unwrap_or(&tb);
+    assert!(
+        ttext.starts_with("restricted: you are timed out until"),
+        "timed-out member must carry the `restricted:` wire contract, got {ttext:?}"
+    );
+
+    // ── Unrestricted member: DM_OPEN still succeeds and mutates. ──
+    let (gs, gb) = post_event(&good, &dm_open(&good, &owner)).await;
+    println!("[clean] DM_OPEN -> {gs} {gb}");
+    assert_eq!(gs, 200, "unrestricted member must be unaffected: {gb}");
+
+    // ── Mutation defense-in-depth: the DM_OPEN attempt creates a DM only for
+    // the unrestricted member. Absolute counts include the ban/timeout notice
+    // DMs, so compare against the pre-attempt snapshots. ──
+    assert_eq!(
+        channel_membership_count(&p, cid, &banned).await,
+        before_banned,
+        "banned member's DM_OPEN must create no DM"
+    );
+    assert_eq!(
+        channel_membership_count(&p, cid, &timed_out).await,
+        before_timed_out,
+        "timed-out member's DM_OPEN must create no DM"
+    );
+    assert!(
+        channel_membership_count(&p, cid, &good).await > before_good,
+        "unrestricted member's DM must actually be created"
+    );
+
+    println!("\nALL INVARIANTS HELD");
+}
+
+fn chrono_now() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs() as i64)
+        .unwrap_or(0)
+}