mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
feat(relay): add atomic create-only community provisioning (#1689)
Signed-off-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 <d12721ccebd9506456c1425125573c09f795d4e2a1f6543f0cd8405dc2575bfe@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 <d12721ccebd9506456c1425125573c09f795d4e2a1f6543f0cd8405dc2575bfe@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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
|
||||
@@ -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:
|
||||
@@ -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)`.
|
||||
@@ -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>> {
|
||||
|
||||
@@ -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 {
|
||||
@@ -457,6 +466,72 @@ impl Db {
|
||||
})
|
||||
}
|
||||
|
||||
/// Atomically creates a community and its initial owner.
|
||||
///
|
||||
/// Returns the existing row when the normalized host already has the same
|
||||
/// current owner, making ambiguous create retries naturally idempotent.
|
||||
/// Returns `None` when the host exists with a different (or missing) owner.
|
||||
/// The initial host and owner inserts share one transaction, so callers
|
||||
/// never observe a partially provisioned community or rotate an owner.
|
||||
pub async fn create_community_with_owner(
|
||||
&self,
|
||||
normalized_host: &str,
|
||||
owner_pubkey: &str,
|
||||
) -> Result<Option<CreatedCommunityRecord>> {
|
||||
let owner_pubkey = owner_pubkey.to_ascii_lowercase();
|
||||
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 (id, host) = if let Some(row) = row {
|
||||
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)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
(id, host)
|
||||
} else {
|
||||
let existing = sqlx::query(
|
||||
r#"
|
||||
SELECT c.id, c.host
|
||||
FROM communities c
|
||||
JOIN relay_members rm ON rm.community_id = c.id
|
||||
WHERE lower(c.host) = lower($1)
|
||||
AND lower(rm.pubkey) = lower($2)
|
||||
AND rm.role = 'owner'
|
||||
"#,
|
||||
)
|
||||
.bind(normalized_host)
|
||||
.bind(&owner_pubkey)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(existing) = existing else {
|
||||
tx.rollback().await?;
|
||||
return Ok(None);
|
||||
};
|
||||
(existing.try_get("id")?, existing.try_get("host")?)
|
||||
};
|
||||
|
||||
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
|
||||
@@ -2925,6 +3000,82 @@ 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 retry = db
|
||||
.create_community_with_owner(&host.to_ascii_uppercase(), owner)
|
||||
.await
|
||||
.expect("same-owner retry")
|
||||
.expect("existing same-owner community");
|
||||
assert_eq!(retry, created, "retry returns the original row");
|
||||
|
||||
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())]);
|
||||
|
||||
db.bootstrap_owner(created.id, other)
|
||||
.await
|
||||
.expect("rotate owner");
|
||||
let post_rotation_retry = db
|
||||
.create_community_with_owner(&host, owner)
|
||||
.await
|
||||
.expect("post-rotation retry");
|
||||
assert!(post_rotation_retry.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn concurrent_same_owner_create_returns_the_winning_row_to_both_callers() {
|
||||
let db = setup_db().await;
|
||||
let host = format!("concurrent-create-{}.example", Uuid::new_v4().simple());
|
||||
let owner = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
|
||||
|
||||
let (first, second) = tokio::join!(
|
||||
db.create_community_with_owner(&host, owner),
|
||||
db.create_community_with_owner(&host, owner),
|
||||
);
|
||||
let first = first
|
||||
.expect("first concurrent create")
|
||||
.expect("first result");
|
||||
let second = second
|
||||
.expect("second concurrent create")
|
||||
.expect("second result");
|
||||
|
||||
assert_eq!(first, second, "conflict loser re-reads the winning row");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn ensure_configured_community_reports_insert_winner() {
|
||||
|
||||
@@ -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.
|
||||
@@ -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>>
|
||||
@@ -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
|
||||
@@ -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"
|
||||
);
|
||||
@@ -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"
|
||||
);
|
||||
|
||||
@@ -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>> {
|
||||
|
||||
@@ -33,8 +33,11 @@ pub struct CommunityAvailabilityQuery {
|
||||
host: String,
|
||||
}
|
||||
|
||||
/// Shared operator auth prelude: bind an ingress host for NIP-98 URL/replay
|
||||
/// scoping, verify the signed request, then gate on `RELAY_OPERATOR_PUBKEYS`.
|
||||
const OPERATOR_REPLAY_SCOPE: &str = "operator-management";
|
||||
|
||||
/// Shared deployment-global operator auth prelude. The canonical management
|
||||
/// origin and replay namespace are configuration, never tenant registry state
|
||||
/// or an inbound proxy `Host` header.
|
||||
async fn authorize_operator_request(
|
||||
state: &Arc<AppState>,
|
||||
headers: &HeaderMap,
|
||||
@@ -43,27 +46,16 @@ async fn authorize_operator_request(
|
||||
raw_query: Option<&str>,
|
||||
body: Option<&[u8]>,
|
||||
) -> Result<nostr::PublicKey, (StatusCode, Json<Value>)> {
|
||||
// Bind to an existing ingress community only for NIP-98 URL/replay scoping.
|
||||
// This is not a tenant data-plane operation, so do not run relay-membership
|
||||
// checks and do not route through event ingest.
|
||||
let raw_host = headers
|
||||
.get(axum::http::header::HOST)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("");
|
||||
let tenant = crate::tenant::bind_community(&state.db, raw_host)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
api_error(
|
||||
StatusCode::NOT_FOUND,
|
||||
"relay: no community is configured for this host",
|
||||
)
|
||||
})?;
|
||||
|
||||
let origin = state
|
||||
.config
|
||||
.relay_operator_api_origin
|
||||
.as_deref()
|
||||
.ok_or_else(|| internal_error("operator API origin is not configured"))?;
|
||||
let path_with_query = match raw_query {
|
||||
Some(q) if !q.is_empty() => format!("{path}?{q}"),
|
||||
_ => path.to_string(),
|
||||
};
|
||||
let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, &path_with_query);
|
||||
let url = format!("{origin}{path_with_query}");
|
||||
let (pubkey, event_id_bytes) = bridge::verify_bridge_auth_with_options(
|
||||
headers,
|
||||
method,
|
||||
@@ -72,7 +64,7 @@ async fn authorize_operator_request(
|
||||
true, // operator endpoints always require NIP-98; no X-Pubkey dev fallback
|
||||
body.is_some(),
|
||||
)?;
|
||||
bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?;
|
||||
check_operator_replay(state, event_id_bytes).await?;
|
||||
|
||||
let pubkey_hex = pubkey.to_hex();
|
||||
if !state
|
||||
@@ -90,7 +82,40 @@ async fn authorize_operator_request(
|
||||
Ok(pubkey)
|
||||
}
|
||||
|
||||
/// Provision or converge a community host.
|
||||
async fn check_operator_replay(
|
||||
state: &AppState,
|
||||
event_id_bytes: [u8; 32],
|
||||
) -> Result<(), (StatusCode, Json<Value>)> {
|
||||
let event_id = nostr::EventId::from_byte_array(event_id_bytes);
|
||||
match state
|
||||
.nip98_replay
|
||||
.try_mark_in_scope(
|
||||
OPERATOR_REPLAY_SCOPE,
|
||||
&event_id,
|
||||
buzz_auth::DEFAULT_REPLAY_TTL_SECS,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(true) => Ok(()),
|
||||
Ok(false) => Err(api_error(
|
||||
StatusCode::UNAUTHORIZED,
|
||||
"NIP-98: replay detected",
|
||||
)),
|
||||
Err(error) => {
|
||||
tracing::warn!(
|
||||
scope = OPERATOR_REPLAY_SCOPE,
|
||||
error = %error,
|
||||
"operator NIP-98 replay guard failed; rejecting request fail-closed"
|
||||
);
|
||||
Err(api_error(
|
||||
StatusCode::UNAUTHORIZED,
|
||||
"NIP-98: replay check unavailable",
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a community host and atomically bootstrap its initial owner.
|
||||
///
|
||||
/// `POST /operator/communities`, NIP-98 signed by a pubkey in
|
||||
/// `RELAY_OPERATOR_PUBKEYS`, body:
|
||||
@@ -99,9 +124,8 @@ async fn authorize_operator_request(
|
||||
/// { "host": "acme.communities.buzz.xyz", "initial_owner_pubkey": "<hex>" }
|
||||
/// ```
|
||||
///
|
||||
/// The request is authenticated against the host it arrives on (so NIP-98 `u`
|
||||
/// still binds to the request authority) but it intentionally does not require
|
||||
/// relay membership in that host's community. The operator allowlist is the
|
||||
/// The request is authenticated against `RELAY_OPERATOR_API_ORIGIN` and does
|
||||
/// not bind the inbound host to a tenant. The operator allowlist is the
|
||||
/// authority for this deployment-root control-plane surface.
|
||||
pub async fn provision_community(
|
||||
State(state): State<Arc<AppState>>,
|
||||
@@ -135,6 +159,14 @@ pub async fn provision_community(
|
||||
Err(msg) if msg.starts_with("actor not authorized") => {
|
||||
Err(api_error(StatusCode::FORBIDDEN, &msg))
|
||||
}
|
||||
Err(msg) if msg == "community already exists" => Err(api_error(StatusCode::CONFLICT, &msg)),
|
||||
Err(msg)
|
||||
if msg.starts_with("failed to create community:")
|
||||
|| msg.starts_with("community provisioned but owner bootstrap failed:") =>
|
||||
{
|
||||
tracing::error!(error = %msg, "operator community persistence failed");
|
||||
Err(internal_error("operator community persistence failed"))
|
||||
}
|
||||
Err(msg) => Err(api_error(StatusCode::BAD_REQUEST, &msg)),
|
||||
}
|
||||
}
|
||||
@@ -234,9 +266,9 @@ mod tests {
|
||||
struct AlwaysFreshReplayGuard;
|
||||
|
||||
impl buzz_auth::Nip98ReplayGuard for AlwaysFreshReplayGuard {
|
||||
fn try_mark<'a>(
|
||||
fn try_mark_in_scope<'a>(
|
||||
&'a self,
|
||||
_ctx: &'a buzz_core::TenantContext,
|
||||
_scope: &'a str,
|
||||
_event_id: &'a nostr::EventId,
|
||||
_ttl_secs: u64,
|
||||
) -> std::pin::Pin<
|
||||
@@ -286,7 +318,8 @@ mod tests {
|
||||
let mut config = crate::config::Config::from_env().ok()?;
|
||||
config.database_url = TEST_DB_URL.to_string();
|
||||
config.redis_url = "redis://127.0.0.1:1".to_string();
|
||||
config.relay_url = format!("wss://{INGRESS_HOST}");
|
||||
config.relay_url = "wss://tenant.example".to_string();
|
||||
config.relay_operator_api_origin = Some(format!("http://{INGRESS_HOST}"));
|
||||
config.relay_operator_pubkeys = operator_keys
|
||||
.iter()
|
||||
.map(|keys| keys.public_key().to_hex())
|
||||
@@ -294,7 +327,6 @@ mod tests {
|
||||
|
||||
let pool = sqlx::PgPool::connect(TEST_DB_URL).await.ok()?;
|
||||
let db = buzz_db::Db::from_pool(pool.clone());
|
||||
db.ensure_configured_community(INGRESS_HOST).await.ok()?;
|
||||
|
||||
let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
|
||||
.create_pool(Some(deadpool_redis::Runtime::Tokio1))
|
||||
@@ -347,7 +379,7 @@ mod tests {
|
||||
r#"{{"host":"community-{}.example"}}"#,
|
||||
Uuid::new_v4().simple()
|
||||
);
|
||||
let url = format!("https://{INGRESS_HOST}/operator/communities");
|
||||
let url = format!("http://{INGRESS_HOST}/operator/communities");
|
||||
let auth = nip98_auth_header(&outsider, &url, "POST", Some(body.as_bytes()));
|
||||
|
||||
let response = build_router(state)
|
||||
@@ -378,7 +410,7 @@ mod tests {
|
||||
r#"{{"host":"community-{}.example"}}"#,
|
||||
Uuid::new_v4().simple()
|
||||
);
|
||||
let url = format!("https://{INGRESS_HOST}/operator/communities");
|
||||
let url = format!("http://{INGRESS_HOST}/operator/communities");
|
||||
let auth = nip98_auth_header_without_payload(&operator, &url, "POST");
|
||||
|
||||
let response = build_router(state)
|
||||
@@ -406,6 +438,68 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn unmapped_management_host_can_check_availability() {
|
||||
let operator = Keys::generate();
|
||||
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
|
||||
return;
|
||||
};
|
||||
let host = format!("community-{}.example", Uuid::new_v4().simple());
|
||||
let query = format!("host={host}");
|
||||
let url = format!("http://{INGRESS_HOST}/operator/communities/availability?{query}");
|
||||
let auth = nip98_auth_header(&operator, &url, "GET", None);
|
||||
|
||||
let response = build_router(state)
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri(format!("/operator/communities/availability?{query}"))
|
||||
.header(header::HOST, INGRESS_HOST)
|
||||
.header(header::AUTHORIZATION, auth)
|
||||
.body(Body::empty())
|
||||
.expect("request"),
|
||||
)
|
||||
.await
|
||||
.expect("response");
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let json = read_json(response).await;
|
||||
assert_eq!(json.get("available").and_then(Value::as_bool), Some(true));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn unmapped_management_host_can_list_owned_communities() {
|
||||
let operator = Keys::generate();
|
||||
let owner = Keys::generate();
|
||||
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
|
||||
return;
|
||||
};
|
||||
let owner_hex = owner.public_key().to_hex();
|
||||
let query = format!("owner_pubkey={owner_hex}");
|
||||
let url = format!("http://{INGRESS_HOST}/operator/communities?{query}");
|
||||
let auth = nip98_auth_header(&operator, &url, "GET", None);
|
||||
|
||||
let response = build_router(state)
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri(format!("/operator/communities?{query}"))
|
||||
.header(header::HOST, INGRESS_HOST)
|
||||
.header(header::AUTHORIZATION, auth)
|
||||
.body(Body::empty())
|
||||
.expect("request"),
|
||||
)
|
||||
.await
|
||||
.expect("response");
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let json = read_json(response).await;
|
||||
assert_eq!(
|
||||
json.get("owner_pubkey").and_then(Value::as_str),
|
||||
Some(owner_hex.as_str())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn happy_path_create_returns_created_and_bootstraps_owner() {
|
||||
@@ -420,7 +514,7 @@ mod tests {
|
||||
"initial_owner_pubkey": owner.public_key().to_hex(),
|
||||
})
|
||||
.to_string();
|
||||
let url = format!("https://{INGRESS_HOST}/operator/communities");
|
||||
let url = format!("http://{INGRESS_HOST}/operator/communities");
|
||||
let auth = nip98_auth_header(&operator, &url, "POST", Some(body.as_bytes()));
|
||||
|
||||
let response = build_router(state.clone())
|
||||
|
||||
@@ -98,12 +98,20 @@ pub struct Config {
|
||||
/// with the `owner` role on first startup.
|
||||
pub relay_owner_pubkey: Option<String>,
|
||||
|
||||
/// Canonical HTTP origin of the deployment-global operator API.
|
||||
///
|
||||
/// Every operator NIP-98 `u` tag is verified against this origin, independent
|
||||
/// of the inbound HTTP `Host` header and tenant registry. Required when
|
||||
/// `RELAY_OPERATOR_PUBKEYS` is non-empty. Set via `RELAY_OPERATOR_API_ORIGIN`
|
||||
/// as an `http://` or `https://` origin with no path, query, or fragment.
|
||||
pub relay_operator_api_origin: Option<String>,
|
||||
|
||||
/// Deployment-level relay operator pubkeys allowed to use the
|
||||
/// `POST /operator/communities` provisioning endpoint.
|
||||
/// `/operator/communities` management endpoints.
|
||||
///
|
||||
/// Unlike `relay_owner_pubkey` (a role *within* the deployment community),
|
||||
/// operators span tenants: they may create new communities and rotate owners
|
||||
/// via the operator endpoint, but hold no implicit tenant membership row.
|
||||
/// operators span tenants: they may create new communities and bootstrap
|
||||
/// initial owners, but hold no implicit tenant membership row.
|
||||
/// Empty (the default) disables community provisioning entirely — fail closed.
|
||||
///
|
||||
/// Set via `RELAY_OPERATOR_PUBKEYS` as a comma-separated list of 64-char
|
||||
@@ -176,6 +184,27 @@ fn parse_bind_addr(raw: &str) -> Result<SocketAddr, ConfigError> {
|
||||
.map_err(|e| ConfigError::InvalidBindAddr(e.to_string()))
|
||||
}
|
||||
|
||||
fn parse_operator_api_origin(raw: &str) -> Result<String, ConfigError> {
|
||||
let raw = raw.trim();
|
||||
let url = url::Url::parse(raw).map_err(|e| {
|
||||
ConfigError::InvalidValue(format!("RELAY_OPERATOR_API_ORIGIN is not a valid URL: {e}"))
|
||||
})?;
|
||||
if !matches!(url.scheme(), "http" | "https")
|
||||
|| url.host().is_none()
|
||||
|| !url.username().is_empty()
|
||||
|| url.password().is_some()
|
||||
|| url.path() != "/"
|
||||
|| url.query().is_some()
|
||||
|| url.fragment().is_some()
|
||||
{
|
||||
return Err(ConfigError::InvalidValue(
|
||||
"RELAY_OPERATOR_API_ORIGIN must be an http(s) origin with no credentials, path, query, or fragment"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
Ok(raw.trim_end_matches('/').to_string())
|
||||
}
|
||||
|
||||
fn ensure_git_repo_path(
|
||||
raw: impl Into<std::path::PathBuf>,
|
||||
) -> Result<std::path::PathBuf, ConfigError> {
|
||||
@@ -278,6 +307,12 @@ impl Config {
|
||||
// pubkeys. Unlike RELAY_OWNER_PUBKEY (warn-and-ignore), an invalid
|
||||
// entry here is a hard config error: silently dropping an operator
|
||||
// pubkey would silently disable provisioning for that operator.
|
||||
let relay_operator_api_origin = std::env::var("RELAY_OPERATOR_API_ORIGIN")
|
||||
.ok()
|
||||
.filter(|raw| !raw.trim().is_empty())
|
||||
.map(|raw| parse_operator_api_origin(&raw))
|
||||
.transpose()?;
|
||||
|
||||
let relay_operator_pubkeys = match std::env::var("RELAY_OPERATOR_PUBKEYS") {
|
||||
Ok(raw) => {
|
||||
let mut pubkeys = Vec::new();
|
||||
@@ -300,6 +335,12 @@ impl Config {
|
||||
}
|
||||
Err(_) => Vec::new(),
|
||||
};
|
||||
if !relay_operator_pubkeys.is_empty() && relay_operator_api_origin.is_none() {
|
||||
return Err(ConfigError::InvalidValue(
|
||||
"RELAY_OPERATOR_API_ORIGIN is required when RELAY_OPERATOR_PUBKEYS is configured"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
let auth = buzz_auth::AuthConfig::default();
|
||||
|
||||
@@ -483,6 +524,7 @@ impl Config {
|
||||
require_relay_membership,
|
||||
huddle_audio_available,
|
||||
relay_owner_pubkey,
|
||||
relay_operator_api_origin,
|
||||
relay_operator_pubkeys,
|
||||
allow_nip_oa_auth,
|
||||
media,
|
||||
@@ -554,8 +596,13 @@ mod tests {
|
||||
"RELAY_OPERATOR_PUBKEYS",
|
||||
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA,bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb,aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
);
|
||||
std::env::set_var(
|
||||
"RELAY_OPERATOR_API_ORIGIN",
|
||||
"http://buzz.mesh.bb-production.com",
|
||||
);
|
||||
let config = Config::from_env().expect("config");
|
||||
std::env::remove_var("RELAY_OPERATOR_PUBKEYS");
|
||||
std::env::remove_var("RELAY_OPERATOR_API_ORIGIN");
|
||||
|
||||
assert_eq!(
|
||||
config.relay_operator_pubkeys,
|
||||
@@ -579,6 +626,36 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn relay_operator_pubkeys_require_api_origin() {
|
||||
let _guard = ENV_MUTEX.lock().unwrap();
|
||||
std::env::set_var(
|
||||
"RELAY_OPERATOR_PUBKEYS",
|
||||
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
);
|
||||
std::env::remove_var("RELAY_OPERATOR_API_ORIGIN");
|
||||
let result = Config::from_env();
|
||||
std::env::remove_var("RELAY_OPERATOR_PUBKEYS");
|
||||
|
||||
assert!(matches!(
|
||||
result,
|
||||
Err(ConfigError::InvalidValue(ref msg)) if msg.contains("RELAY_OPERATOR_API_ORIGIN is required")
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn relay_operator_api_origin_rejects_paths() {
|
||||
let _guard = ENV_MUTEX.lock().unwrap();
|
||||
std::env::set_var("RELAY_OPERATOR_API_ORIGIN", "https://buzz.example/operator");
|
||||
let result = Config::from_env();
|
||||
std::env::remove_var("RELAY_OPERATOR_API_ORIGIN");
|
||||
|
||||
assert!(matches!(
|
||||
result,
|
||||
Err(ConfigError::InvalidValue(ref msg)) if msg.contains("must be an http(s) origin")
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn huddle_audio_available_can_be_disabled_for_horizontal_scaling() {
|
||||
let _guard = ENV_MUTEX.lock().unwrap();
|
||||
|
||||
@@ -47,6 +47,10 @@ pub struct ProvisionCommunityRequest {
|
||||
/// rotates the owner through the same bootstrap path used at startup.
|
||||
#[serde(default)]
|
||||
pub initial_owner_pubkey: Option<String>,
|
||||
/// When true, atomically create the host and owner and reject an existing
|
||||
/// host instead of converging or rotating ownership.
|
||||
#[serde(default)]
|
||||
pub create_only: bool,
|
||||
}
|
||||
|
||||
/// JSON response from `POST /operator/communities`.
|
||||
@@ -245,10 +249,35 @@ pub async fn provision_community(
|
||||
})
|
||||
.transpose()?;
|
||||
|
||||
// Same idempotent upsert as the startup seed — creating a community is an
|
||||
// INSERT, never DDL (docs/multi-tenant-relay.md §System Model). The helper
|
||||
// returns whether this call inserted the row, so concurrent provisioners do
|
||||
// not both report `created` after an upsert conflict.
|
||||
if request.create_only {
|
||||
let owner_hex = initial_owner.as_deref().ok_or_else(|| {
|
||||
"initial_owner_pubkey is required when create_only is true".to_string()
|
||||
})?;
|
||||
let record = state
|
||||
.db
|
||||
.create_community_with_owner(&request.host, owner_hex)
|
||||
.await
|
||||
.map_err(|e| format!("failed to create community: {e}"))?
|
||||
.ok_or_else(|| "community already exists".to_string())?;
|
||||
|
||||
info!(
|
||||
operator = %operator_hex,
|
||||
community = %record.id,
|
||||
host = %record.host,
|
||||
owner = %owner_hex,
|
||||
"community created via operator endpoint"
|
||||
);
|
||||
return Ok(ProvisionCommunityResponse {
|
||||
community_id: record.id.to_string(),
|
||||
host: record.host,
|
||||
status: "created",
|
||||
owner_pubkey: initial_owner,
|
||||
});
|
||||
}
|
||||
|
||||
// Legacy convergence mode remains available to deployment operators and
|
||||
// startup tooling. Clients provisioning on behalf of end users must use
|
||||
// create_only so an existing owner can never be rotated by a create race.
|
||||
let record = state
|
||||
.db
|
||||
.ensure_configured_community(&request.host)
|
||||
|
||||
Reference in New Issue
Block a user