Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion crates/buzz-auth/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,8 @@ pub use error::AuthError;
pub use nip42::{generate_challenge, verify_nip42_event};
pub use nip98::verify_nip98_event;
pub use nip98_replay::{
nip98_replay_key, Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS,
nip98_replay_key, nip98_replay_key_for_scope, Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS,
MAX_REPLAY_TTL_SECS,
};
pub use rate_limit::{
ip_rate_limit_key, rate_limit_key, LimitType, RateLimitConfig, RateLimitResult, RateLimiter,
Expand Down
24 changes: 20 additions & 4 deletions crates/buzz-auth/src/nip98_replay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,14 @@ pub const MAX_REPLAY_TTL_SECS: u64 = 3600;
/// The production implementation lives in `buzz-pubsub` (Redis `SET NX EX`).
/// A test impl is provided behind `cfg(any(test, feature = "test-utils"))`.
pub trait Nip98ReplayGuard: Send + Sync {
/// Atomically claim `event_id` in an explicit deployment or community scope.
fn try_mark_in_scope<'a>(
&'a self,
scope: &'a str,
event_id: &'a EventId,
ttl_secs: u64,
) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + Send + 'a>>;

/// Atomically claim `event_id` for `ctx`'s community.
///
/// Returns `Ok(true)` when the id is newly inserted (proceed) and
Expand Down Expand Up @@ -89,7 +97,10 @@ pub trait Nip98ReplayGuard: Send + Sync {
ctx: &'a TenantContext,
event_id: &'a EventId,
ttl_secs: u64,
) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + Send + 'a>>;
) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + Send + 'a>> {
let scope = ctx.community().to_string();
Box::pin(async move { self.try_mark_in_scope(&scope, event_id, ttl_secs).await })
}
}

/// Redis key for a NIP-98 replay marker:
Expand All @@ -101,7 +112,12 @@ pub trait Nip98ReplayGuard: Send + Sync {
/// isolation: a same-id replay across communities must consult two distinct
/// seen-set rows, not one shared row.
pub fn nip98_replay_key(ctx: &TenantContext, event_id: &EventId) -> String {
format!("buzz:{}:nip98:{}", ctx.community(), event_id.to_hex())
nip98_replay_key_for_scope(&ctx.community().to_string(), event_id)
}

/// Redis key for a NIP-98 replay marker in an explicit trusted scope.
pub fn nip98_replay_key_for_scope(scope: &str, event_id: &EventId) -> String {
format!("buzz:{scope}:nip98:{}", event_id.to_hex())
}

/// Always-fresh seen-set for unit tests — every `try_mark` returns `Ok(true)`.
Expand All @@ -112,9 +128,9 @@ pub struct AlwaysFreshReplayGuard;

#[cfg(any(test, feature = "test-utils"))]
impl Nip98ReplayGuard for AlwaysFreshReplayGuard {
fn try_mark<'a>(
fn try_mark_in_scope<'a>(
&'a self,
_ctx: &'a TenantContext,
_scope: &'a str,
_event_id: &'a EventId,
_ttl_secs: u64,
) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + Send + 'a>> {
Expand Down
92 changes: 92 additions & 0 deletions crates/buzz-db/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,15 @@ pub struct EnsuredCommunityRecord {
pub created: bool,
}

/// Community row returned by an atomic create-with-owner operation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatedCommunityRecord {
/// Stable server-resolved community id.
pub id: CommunityId,
/// Normalized host stored for the community.
pub host: String,
}

/// Community row returned by operator-plane ownership reads.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OwnedCommunityRecord {
Expand Down Expand Up @@ -457,6 +466,50 @@ impl Db {
})
}

/// Atomically creates a community and its initial owner.
///
/// Returns `None` when the normalized host already exists. The host insert
/// and owner insert share one transaction, so callers never observe a
/// partially provisioned community and cannot rotate an existing owner.
pub async fn create_community_with_owner(
&self,
normalized_host: &str,
owner_pubkey: &str,
) -> Result<Option<CreatedCommunityRecord>> {
let mut tx = self.pool.begin().await?;
let row = sqlx::query(
r#"
INSERT INTO communities (host)
VALUES ($1)
ON CONFLICT (lower(host)) DO NOTHING
RETURNING id, host
"#,
)
.bind(normalized_host)
.fetch_optional(&mut *tx)
.await?;

let Some(row) = row else {
tx.rollback().await?;
return Ok(None);
};
let id: Uuid = row.try_get("id")?;
let host: String = row.try_get("host")?;
sqlx::query(
"INSERT INTO relay_members (community_id, pubkey, role, added_by) VALUES ($1, $2, 'owner', NULL)",
)
.bind(id)
.bind(owner_pubkey.to_ascii_lowercase())
.execute(&mut *tx)
.await?;
tx.commit().await?;

Ok(Some(CreatedCommunityRecord {
id: CommunityId::from_uuid(id),
host,
}))
}

/// Returns the community that owns a channel, if the channel exists.
///
/// Internal relay producers use this to derive tenant context from the row
Expand Down Expand Up @@ -2925,6 +2978,45 @@ mod tests {
assert_eq!(found.id, CommunityId::from_uuid(id));
}

#[tokio::test]
#[ignore = "requires Postgres"]
async fn create_community_with_owner_is_atomic_and_create_only() {
let db = setup_db().await;
let host = format!("create-only-{}.example", Uuid::new_v4().simple());
let owner = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let other = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";

let created = db
.create_community_with_owner(&host, owner)
.await
.expect("create community")
.expect("new host");
assert_eq!(created.host, host);
let owner_role: Option<String> = sqlx::query_scalar(
"SELECT role FROM relay_members WHERE community_id = $1 AND pubkey = $2",
)
.bind(created.id.as_uuid())
.bind(owner)
.fetch_optional(&db.pool)
.await
.expect("owner role");
assert_eq!(owner_role.as_deref(), Some("owner"));

let collision = db
.create_community_with_owner(&host, other)
.await
.expect("collision result");
assert!(collision.is_none());
let roles: Vec<(String, String)> = sqlx::query_as(
"SELECT pubkey, role FROM relay_members WHERE community_id = $1 ORDER BY pubkey",
)
.bind(created.id.as_uuid())
.fetch_all(&db.pool)
.await
.expect("community roles");
assert_eq!(roles, vec![(owner.to_string(), "owner".to_string())]);
}

#[tokio::test]
#[ignore = "requires Postgres"]
async fn ensure_configured_community_reports_insert_winner() {
Expand Down
15 changes: 7 additions & 8 deletions crates/buzz-pubsub/src/nip98_replay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,9 @@
use buzz_auth::{
error::AuthError,
nip98_replay::{
nip98_replay_key, Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS,
nip98_replay_key_for_scope, Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS,
},
};
use buzz_core::TenantContext;
use nostr::EventId;

/// Redis-backed NIP-98 replay seen-set.
Expand All @@ -33,9 +32,9 @@ impl RedisNip98ReplayGuard {
}

impl Nip98ReplayGuard for RedisNip98ReplayGuard {
fn try_mark<'a>(
fn try_mark_in_scope<'a>(
&'a self,
ctx: &'a TenantContext,
scope: &'a str,
event_id: &'a EventId,
ttl_secs: u64,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<bool, AuthError>> + Send + 'a>>
Expand All @@ -52,14 +51,14 @@ impl Nip98ReplayGuard for RedisNip98ReplayGuard {
// Structured field for ops; the user-facing AuthError stays a
// bounded category string.
tracing::warn!(
community = %ctx.community(),
scope = %scope,
error = %e,
"nip98 replay: redis pool acquire failed — caller MUST fail closed"
);
AuthError::Internal(format!("Redis pool: {e}"))
})?;

let key = nip98_replay_key(ctx, event_id);
let key = nip98_replay_key_for_scope(scope, event_id);

// SET key 1 NX EX <ttl>. redis-rs typed return: Some("OK") on first
// claim, None on existing key. Any other value would be a Redis-side
Expand All @@ -74,7 +73,7 @@ impl Nip98ReplayGuard for RedisNip98ReplayGuard {
.await
.map_err(|e| {
tracing::warn!(
community = %ctx.community(),
scope = %scope,
error = %e,
"nip98 replay: redis SET NX EX failed — caller MUST fail closed"
);
Expand All @@ -86,7 +85,7 @@ impl Nip98ReplayGuard for RedisNip98ReplayGuard {
None => Ok(false),
Some(other) => {
tracing::error!(
community = %ctx.community(),
scope = %scope,
reply = %other,
"nip98 replay: redis SET NX EX returned an unexpected reply — investigate"
);
Expand Down
4 changes: 2 additions & 2 deletions crates/buzz-relay/src/api/bridge.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2011,9 +2011,9 @@ mod tests {

struct AlwaysErrGuard;
impl Nip98ReplayGuard for AlwaysErrGuard {
fn try_mark<'a>(
fn try_mark_in_scope<'a>(
&'a self,
_ctx: &'a buzz_core::TenantContext,
_scope: &'a str,
_event_id: &'a EventId,
_ttl_secs: u64,
) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + Send + 'a>> {
Expand Down
Loading
Loading