feat(relay): add operator community provisioning (#1657)

Signed-off-by: npub1qvn3cujt28pg06ehlstrxyz6ayzp06t4uc7r566vxwwgrv24hglq9zju0n <03271c724b51c287eb37fc1633105ae90417e975e63c3a6b4c339c81b155ba3e@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1qvn3cujt28pg06ehlstrxyz6ayzp06t4uc7r566vxwwgrv24hglq9zju0n <03271c724b51c287eb37fc1633105ae90417e975e63c3a6b4c339c81b155ba3e@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Bradley Axen
2026-07-08 16:56:06 -07:00
committed by GitHub
parent fc8a17445a
commit e0f76b0e9c
9 changed files with 1108 additions and 15 deletions

View File

@@ -151,7 +151,7 @@ pub struct DbPoolStats {
/// Configuration for the Postgres connection pool.
#[derive(Debug, Clone)]
pub struct DbConfig {
/// Postgres connection URL (e.g. `postgres://user:pass@host/db`).
/// Postgres connection URL (usually sourced from `DATABASE_URL`).
pub database_url: String,
/// Maximum number of connections in the pool.
pub max_connections: u32,
@@ -171,7 +171,7 @@ impl Default for DbConfig {
/// At 20 main + 5 audit = 25/pod, four relay pods fit within the PG limit.
fn default() -> Self {
Self {
database_url: "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string(),
database_url: "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string(), // sadscan:disable np.postgres.1
max_connections: 20,
min_connections: 2,
acquire_timeout_secs: 3,
@@ -190,6 +190,28 @@ pub struct CommunityRecord {
pub host: String,
}
/// Community row returned by idempotent community ensure/create operations.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnsuredCommunityRecord {
/// Stable server-resolved community id.
pub id: CommunityId,
/// Normalized host that maps to this community.
pub host: String,
/// True only when this call inserted the `communities` row.
pub created: bool,
}
/// Community row returned by operator-plane ownership reads.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OwnedCommunityRecord {
/// Stable server-resolved community id.
pub id: CommunityId,
/// Normalized host that maps to this community.
pub host: String,
/// When the community row was created.
pub created_at: DateTime<Utc>,
}
/// Token summary returned by [`Db::list_active_tokens`].
#[derive(Debug, Clone)]
pub struct TokenSummary {
@@ -294,6 +316,43 @@ impl Db {
.transpose()
}
/// Lists communities where `owner_pubkey` currently holds the `owner` role.
///
/// This is an operator-plane helper, not a tenant-scoped data-plane read:
/// callers must gate it on deployment-level operator auth before exposing it.
pub async fn list_communities_owned_by(
&self,
owner_pubkey: &str,
) -> Result<Vec<OwnedCommunityRecord>> {
let owner_pubkey = owner_pubkey.to_ascii_lowercase();
let rows = sqlx::query(
r#"
SELECT c.id, c.host, c.created_at
FROM communities c
JOIN relay_members rm ON rm.community_id = c.id
WHERE rm.pubkey = $1
AND rm.role = 'owner'
ORDER BY c.created_at ASC, c.host ASC
"#,
)
.bind(owner_pubkey)
.fetch_all(&self.pool)
.await?;
rows.into_iter()
.map(|row| {
let id: Uuid = row.try_get("id")?;
let host: String = row.try_get("host")?;
let created_at: DateTime<Utc> = row.try_get("created_at")?;
Ok(OwnedCommunityRecord {
id: CommunityId::from_uuid(id),
host,
created_at,
})
})
.collect()
}
/// Returns the normalized host mapped to a community id, if the community
/// exists.
///
@@ -374,13 +433,13 @@ impl Db {
pub async fn ensure_configured_community(
&self,
normalized_host: &str,
) -> Result<CommunityRecord> {
) -> Result<EnsuredCommunityRecord> {
let row = sqlx::query(
r#"
INSERT INTO communities (host)
VALUES ($1)
ON CONFLICT (lower(host)) DO UPDATE SET host = EXCLUDED.host
RETURNING id, host
ON CONFLICT (lower(host)) DO UPDATE SET host = communities.host
RETURNING id, host, (xmax = 0) AS created
"#,
)
.bind(normalized_host)
@@ -389,10 +448,12 @@ impl Db {
let id: Uuid = row.try_get("id")?;
let host: String = row.try_get("host")?;
let created: bool = row.try_get("created")?;
Ok(CommunityRecord {
Ok(EnsuredCommunityRecord {
id: CommunityId::from_uuid(id),
host,
created,
})
}
@@ -2864,6 +2925,57 @@ mod tests {
assert_eq!(found.id, CommunityId::from_uuid(id));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn ensure_configured_community_reports_insert_winner() {
let db = setup_db().await;
let host = format!("ensure-community-{}.example", Uuid::new_v4().simple());
let first = db
.ensure_configured_community(&host)
.await
.expect("first ensure");
assert!(first.created, "first ensure should report created");
assert_eq!(first.host, host);
let second = db
.ensure_configured_community(&host)
.await
.expect("second ensure");
assert!(!second.created, "second ensure should report existed");
assert_eq!(second.id, first.id);
assert_eq!(second.host, host);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn list_communities_owned_by_returns_only_owner_rows() {
let db = setup_db().await;
let community_a = CommunityId::from_uuid(make_community(&db.pool).await);
let community_b = CommunityId::from_uuid(make_community(&db.pool).await);
let community_c = CommunityId::from_uuid(make_community(&db.pool).await);
let owner = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let other = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
db.bootstrap_owner(community_a, owner)
.await
.expect("owner A");
db.bootstrap_owner(community_b, other)
.await
.expect("other owner B");
db.add_relay_member(community_c, owner, "admin", None)
.await
.expect("admin C");
let owned = db
.list_communities_owned_by(owner)
.await
.expect("list owned communities");
assert_eq!(owned.len(), 1);
assert_eq!(owned[0].id, community_a);
}
async fn insert_channel(pool: &PgPool, community_id: Uuid, channel_id: Uuid) {
let creator: Vec<u8> = vec![0u8; 32];
sqlx::query(

View File

@@ -25,12 +25,23 @@ use super::{api_error, internal_error, not_found};
///
/// Returns the authenticated public key and an event ID for replay detection.
/// For X-Pubkey dev mode, the event ID is a zero hash (no replay concern).
fn verify_bridge_auth(
pub(crate) fn verify_bridge_auth(
headers: &HeaderMap,
method: &str,
url: &str,
body: Option<&[u8]>,
require_auth_token: bool,
) -> Result<(nostr::PublicKey, [u8; 32]), (StatusCode, Json<Value>)> {
verify_bridge_auth_with_options(headers, method, url, body, require_auth_token, false)
}
pub(crate) fn verify_bridge_auth_with_options(
headers: &HeaderMap,
method: &str,
url: &str,
body: Option<&[u8]>,
require_auth_token: bool,
require_payload: bool,
) -> Result<(nostr::PublicKey, [u8; 32]), (StatusCode, Json<Value>)> {
// Try NIP-98 first (Authorization: Nostr <base64>)
if let Some(auth_str) = headers
@@ -51,6 +62,18 @@ fn verify_bridge_auth(
.map_err(|_| api_error(StatusCode::UNAUTHORIZED, "invalid NIP-98 event JSON"))?;
let event_id_bytes = event.id.to_bytes();
if require_payload
&& !event
.tags
.iter()
.any(|tag| tag.kind() == nostr::TagKind::Payload)
{
return Err(api_error(
StatusCode::UNAUTHORIZED,
"NIP-98: missing payload tag",
));
}
let pubkey = buzz_auth::verify_nip98_event(&event_json, url, method, body)
.map_err(|e| api_error(StatusCode::UNAUTHORIZED, &format!("NIP-98: {e}")))?;
@@ -76,7 +99,7 @@ fn verify_bridge_auth(
/// `AppState`, not process-local memory. Any Redis/guard error fails closed:
/// without the shared `SET NX EX` proof, a stateless worker cannot admit the
/// NIP-98 request safely.
async fn check_nip98_replay(
pub(crate) async fn check_nip98_replay(
state: &AppState,
tenant: &TenantContext,
event_id_bytes: [u8; 32],
@@ -135,7 +158,11 @@ async fn check_nip98_replay_with_guard(
/// pass and the relay would proceed against the wrong tenant's auth context),
/// and (b) reject every legitimate request whose community host isn't the
/// single configured one. Substituting `tenant.host()` closes both directions.
fn nip98_expected_url(config_relay_url: &str, tenant: &TenantContext, path: &str) -> String {
pub(crate) fn nip98_expected_url(
config_relay_url: &str,
tenant: &TenantContext,
path: &str,
) -> String {
let scheme = if config_relay_url.trim_start().starts_with("wss://") {
"https"
} else {
@@ -560,6 +587,10 @@ pub async fn submit_event(
check_nip98_replay(&state, &tenant, event_id_bytes).await?;
let pubkey_bytes = pubkey.to_bytes().to_vec();
let event: nostr::Event = serde_json::from_slice(&body)
.map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid event JSON: {e}")))?;
let kind_u32 = buzz_core::kind::event_kind_u32(&event);
// Enforce relay membership (with NIP-OA fallback via x-auth-tag header).
let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok());
super::relay_members::enforce_relay_membership(
@@ -570,16 +601,12 @@ pub async fn submit_event(
)
.await?;
let event: nostr::Event = serde_json::from_slice(&body)
.map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid event JSON: {e}")))?;
// Mesh signaling kinds (24620 status report, 24621 connect request) are
// ephemeral and deliberately absent from ingest_event's per-kind allowlist.
// The desktop's Rust coordinator publishes them via this bridge, so route
// them to the mesh handlers — the HTTP twin of the WS door's special-casing
// in handlers::event. Membership was enforced above; the handlers re-check
// it fail-closed.
let kind_u32 = buzz_core::kind::event_kind_u32(&event);
if kind_u32 == buzz_core::kind::KIND_MESH_STATUS_REPORT
|| kind_u32 == buzz_core::kind::KIND_MESH_CONNECT_REQUEST
{
@@ -2097,6 +2124,34 @@ mod tests {
);
}
#[test]
fn verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints() {
let keys = Keys::generate();
let signed_url = "https://host-a.example/operator/communities";
let event_json = build_nip98_event_json(&keys, signed_url, "POST");
let headers = nip98_auth_headers(&event_json);
let (status, body) = verify_bridge_auth_with_options(
&headers,
"POST",
signed_url,
Some(br#"{"host":"created.example"}"#),
true,
true,
)
.expect_err("body-bearing operator requests must require a payload tag");
assert_eq!(status, StatusCode::UNAUTHORIZED);
let msg = body
.get("error")
.and_then(|v| v.as_str())
.unwrap_or_default();
assert!(
msg.contains("missing payload tag"),
"rejection should explain the payload binding failure; got body = {body:?}"
);
}
/// Positive control for the cross-host test: a NIP-98 event signed for
/// host A MUST be accepted at a request whose tenant resolved to host A.
/// Without this, the cross-host test could be passing vacuously (e.g. if

View File

@@ -5,6 +5,7 @@ pub mod events;
pub mod git;
pub mod media;
pub mod nip05;
pub mod operator;
// Re-export imeta helpers used by ingest pipeline.
pub use crate::handlers::imeta::{validate_imeta_tags, verify_imeta_blobs};

View File

@@ -0,0 +1,461 @@
//! Deployment-operator HTTP APIs.
//!
//! These routes are outside the Nostr event data plane. They still use NIP-98
//! request signing and replay protection, but they do not run through event
//! ingest, relay membership, channel scoping, storage, or fan-out.
use std::sync::Arc;
use axum::{
extract::{Query, RawQuery, State},
http::{HeaderMap, StatusCode},
response::Json,
};
use serde::Deserialize;
use serde_json::Value;
use crate::handlers::community_provisioning::{
normalize_candidate_host, validate_pubkey_hex, ProvisionCommunityRequest,
};
use crate::state::AppState;
use super::{api_error, bridge, internal_error};
/// Query parameters for `GET /operator/communities`.
#[derive(Debug, Deserialize)]
pub struct ListCommunitiesQuery {
owner_pubkey: String,
}
/// Query parameters for `GET /operator/communities/availability`.
#[derive(Debug, Deserialize)]
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`.
async fn authorize_operator_request(
state: &Arc<AppState>,
headers: &HeaderMap,
method: &str,
path: &str,
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 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 (pubkey, event_id_bytes) = bridge::verify_bridge_auth_with_options(
headers,
method,
&url,
body,
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?;
let pubkey_hex = pubkey.to_hex();
if !state
.config
.relay_operator_pubkeys
.iter()
.any(|pk| pk == &pubkey_hex)
{
return Err(api_error(
StatusCode::FORBIDDEN,
"actor not authorized: not a relay operator",
));
}
Ok(pubkey)
}
/// Provision or converge a community host.
///
/// `POST /operator/communities`, NIP-98 signed by a pubkey in
/// `RELAY_OPERATOR_PUBKEYS`, body:
///
/// ```json
/// { "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
/// authority for this deployment-root control-plane surface.
pub async fn provision_community(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let pubkey = authorize_operator_request(
&state,
&headers,
"POST",
"/operator/communities",
None,
Some(&body),
)
.await?;
let request: ProvisionCommunityRequest = serde_json::from_slice(&body).map_err(|e| {
api_error(
StatusCode::BAD_REQUEST,
&format!("invalid provision-community JSON: {e}"),
)
})?;
match crate::handlers::community_provisioning::provision_community(&state, &pubkey, request)
.await
{
Ok(response) => Ok(Json(serde_json::to_value(response).map_err(|e| {
tracing::error!("failed to serialize provision-community response: {e}");
internal_error("operator provision response serialization failed")
})?)),
Err(msg) if msg.starts_with("actor not authorized") => {
Err(api_error(StatusCode::FORBIDDEN, &msg))
}
Err(msg) => Err(api_error(StatusCode::BAD_REQUEST, &msg)),
}
}
/// List communities where a pubkey currently holds the `owner` role.
pub async fn list_owned_communities(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
RawQuery(raw_query): RawQuery,
Query(query): Query<ListCommunitiesQuery>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
authorize_operator_request(
&state,
&headers,
"GET",
"/operator/communities",
raw_query.as_deref(),
None,
)
.await?;
let owner_pubkey = validate_pubkey_hex(&query.owner_pubkey).ok_or_else(|| {
api_error(
StatusCode::BAD_REQUEST,
"invalid owner_pubkey: expected 64-char hex pubkey",
)
})?;
let rows = state
.db
.list_communities_owned_by(&owner_pubkey)
.await
.map_err(|e| internal_error(&format!("list owned communities: {e}")))?;
Ok(Json(serde_json::json!({
"owner_pubkey": owner_pubkey,
"communities": rows.into_iter().map(|row| serde_json::json!({
"community_id": row.id.to_string(),
"host": row.host,
"created_at": row.created_at,
})).collect::<Vec<_>>(),
})))
}
/// Check whether a community host is available, returning the relay-canonical
/// normalized authority used by create.
pub async fn community_availability(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
RawQuery(raw_query): RawQuery,
Query(query): Query<CommunityAvailabilityQuery>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
authorize_operator_request(
&state,
&headers,
"GET",
"/operator/communities/availability",
raw_query.as_deref(),
None,
)
.await?;
let normalized_host = normalize_candidate_host(&query.host)
.map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?;
let existing = state
.db
.lookup_community_by_host(&normalized_host)
.await
.map_err(|e| internal_error(&format!("check community availability: {e}")))?;
Ok(Json(serde_json::json!({
"host": query.host,
"normalized_host": normalized_host,
"available": existing.is_none(),
"community_id": existing.map(|record| record.id.to_string()),
})))
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use axum::{
body::{to_bytes, Body},
http::{header, Request, StatusCode},
};
use base64::Engine;
use nostr::{EventBuilder, Keys, Kind, Tag};
use serde_json::Value;
use sha2::{Digest, Sha256};
use tower::ServiceExt;
use uuid::Uuid;
use crate::router::build_router;
use crate::state::AppState;
struct AlwaysFreshReplayGuard;
impl buzz_auth::Nip98ReplayGuard for AlwaysFreshReplayGuard {
fn try_mark<'a>(
&'a self,
_ctx: &'a buzz_core::TenantContext,
_event_id: &'a nostr::EventId,
_ttl_secs: u64,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<bool, buzz_auth::AuthError>> + Send + 'a>,
> {
Box::pin(async { Ok(true) })
}
}
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1
const INGRESS_HOST: &str = "operator-ingress.example";
fn nip98_auth_header(keys: &Keys, url: &str, method: &str, body: Option<&[u8]>) -> String {
let mut tags = vec![
Tag::parse(["u", url]).expect("u tag"),
Tag::parse(["method", method]).expect("method tag"),
];
if let Some(body) = body {
let hash: [u8; 32] = Sha256::digest(body).into();
let hash_hex = hex::encode(hash);
tags.push(Tag::parse(["payload", hash_hex.as_str()]).expect("payload tag"));
}
let event = EventBuilder::new(Kind::HttpAuth, "")
.tags(tags)
.sign_with_keys(keys)
.expect("sign NIP-98 event");
let event_json = serde_json::to_string(&event).expect("serialize NIP-98 event");
let encoded = base64::engine::general_purpose::STANDARD.encode(event_json.as_bytes());
format!("Nostr {encoded}")
}
fn nip98_auth_header_without_payload(keys: &Keys, url: &str, method: &str) -> String {
let tags = vec![
Tag::parse(["u", url]).expect("u tag"),
Tag::parse(["method", method]).expect("method tag"),
];
let event = EventBuilder::new(Kind::HttpAuth, "")
.tags(tags)
.sign_with_keys(keys)
.expect("sign NIP-98 event");
let event_json = serde_json::to_string(&event).expect("serialize NIP-98 event");
let encoded = base64::engine::general_purpose::STANDARD.encode(event_json.as_bytes());
format!("Nostr {encoded}")
}
async fn operator_test_state(operator_keys: &[Keys]) -> Option<Arc<AppState>> {
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_operator_pubkeys = operator_keys
.iter()
.map(|keys| keys.public_key().to_hex())
.collect();
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))
.ok()?;
let pubsub = Arc::new(
buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
.await
.ok()?,
);
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).ok()?;
let (mut state, _audit_shutdown) = AppState::new(
config,
db,
redis_pool,
audit,
pubsub,
auth,
search,
workflow_engine,
Keys::generate(),
media_storage,
);
state.nip98_replay = Arc::new(AlwaysFreshReplayGuard);
Some(Arc::new(state))
}
async fn read_json(response: axum::response::Response) -> Value {
let bytes = to_bytes(response.into_body(), 1024 * 1024)
.await
.expect("read response body");
serde_json::from_slice(&bytes).expect("response JSON")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn non_allowlisted_operator_key_gets_403() {
let operator = Keys::generate();
let outsider = Keys::generate();
let Some(state) = operator_test_state(&[operator]).await else {
return;
};
let body = format!(
r#"{{"host":"community-{}.example"}}"#,
Uuid::new_v4().simple()
);
let url = format!("https://{INGRESS_HOST}/operator/communities");
let auth = nip98_auth_header(&outsider, &url, "POST", Some(body.as_bytes()));
let response = build_router(state)
.oneshot(
Request::builder()
.method("POST")
.uri("/operator/communities")
.header(header::HOST, INGRESS_HOST)
.header(header::AUTHORIZATION, auth)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(body))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn post_operator_body_requires_payload_tag() {
let operator = Keys::generate();
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
return;
};
let body = format!(
r#"{{"host":"community-{}.example"}}"#,
Uuid::new_v4().simple()
);
let url = format!("https://{INGRESS_HOST}/operator/communities");
let auth = nip98_auth_header_without_payload(&operator, &url, "POST");
let response = build_router(state)
.oneshot(
Request::builder()
.method("POST")
.uri("/operator/communities")
.header(header::HOST, INGRESS_HOST)
.header(header::AUTHORIZATION, auth)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(body))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
let json = read_json(response).await;
assert!(
json.get("error")
.and_then(Value::as_str)
.unwrap_or_default()
.contains("missing payload tag"),
"unexpected response: {json:?}"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn happy_path_create_returns_created_and_bootstraps_owner() {
let operator = Keys::generate();
let owner = 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 body = serde_json::json!({
"host": host,
"initial_owner_pubkey": owner.public_key().to_hex(),
})
.to_string();
let url = format!("https://{INGRESS_HOST}/operator/communities");
let auth = nip98_auth_header(&operator, &url, "POST", Some(body.as_bytes()));
let response = build_router(state.clone())
.oneshot(
Request::builder()
.method("POST")
.uri("/operator/communities")
.header(header::HOST, INGRESS_HOST)
.header(header::AUTHORIZATION, auth)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(body))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let json = read_json(response).await;
assert_eq!(json.get("status").and_then(Value::as_str), Some("created"));
assert_eq!(
json.get("host").and_then(Value::as_str),
Some(host.as_str())
);
let community = state
.db
.lookup_community_by_host(&host)
.await
.expect("lookup community")
.expect("community exists");
let member = state
.db
.get_relay_member(community.id, &owner.public_key().to_hex())
.await
.expect("lookup owner role")
.expect("owner member exists");
assert_eq!(member.role, "owner");
}
}

View File

@@ -98,6 +98,19 @@ pub struct Config {
/// with the `owner` role on first startup.
pub relay_owner_pubkey: Option<String>,
/// Deployment-level relay operator pubkeys allowed to use the
/// `POST /operator/communities` provisioning endpoint.
///
/// 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.
/// Empty (the default) disables community provisioning entirely — fail closed.
///
/// Set via `RELAY_OPERATOR_PUBKEYS` as a comma-separated list of 64-char
/// hex pubkeys. Invalid entries are rejected at startup (config error), not
/// skipped — a typo must not silently disable an operator.
pub relay_operator_pubkeys: Vec<String>,
/// Allow NIP-OA owner attestation for relay membership.
///
/// When `true` and `require_relay_membership` is also `true`, agents
@@ -184,7 +197,7 @@ impl Config {
let bind_addr = parse_bind_addr(&bind_addr_raw)?;
let database_url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); // sadscan:disable np.postgres.1
let redis_url =
std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://localhost:6379".to_string());
@@ -260,6 +273,34 @@ impl Config {
}
});
// Note: intentionally not prefixed with BUZZ_ — same relay-identity
// config family as RELAY_OWNER_PUBKEY. Comma-separated 64-char hex
// 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_pubkeys = match std::env::var("RELAY_OPERATOR_PUBKEYS") {
Ok(raw) => {
let mut pubkeys = Vec::new();
for entry in raw.split(',') {
let entry = entry.trim().to_lowercase();
if entry.is_empty() {
continue;
}
let valid = entry.len() == 64 && entry.chars().all(|c| c.is_ascii_hexdigit());
if !valid {
return Err(ConfigError::InvalidValue(format!(
"RELAY_OPERATOR_PUBKEYS entry is not a valid 64-char hex pubkey: {entry:?}"
)));
}
if !pubkeys.contains(&entry) {
pubkeys.push(entry);
}
}
pubkeys
}
Err(_) => Vec::new(),
};
let auth = buzz_auth::AuthConfig::default();
if !require_auth_token {
@@ -428,6 +469,7 @@ impl Config {
require_relay_membership,
huddle_audio_available,
relay_owner_pubkey,
relay_operator_pubkeys,
allow_nip_oa_auth,
media,
media_max_concurrent_uploads,
@@ -477,6 +519,10 @@ mod tests {
config.relay_owner_pubkey.is_none(),
"relay_owner_pubkey should default to None"
);
assert!(
config.relay_operator_pubkeys.is_empty(),
"relay_operator_pubkeys should default empty (provisioning disabled)"
);
assert!(
!config.allow_nip_oa_auth,
"allow_nip_oa_auth should default to false"
@@ -487,6 +533,38 @@ mod tests {
);
}
#[test]
fn relay_operator_pubkeys_parse_dedupe_and_normalize() {
let _guard = ENV_MUTEX.lock().unwrap();
std::env::set_var(
"RELAY_OPERATOR_PUBKEYS",
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA,bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb,aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
);
let config = Config::from_env().expect("config");
std::env::remove_var("RELAY_OPERATOR_PUBKEYS");
assert_eq!(
config.relay_operator_pubkeys,
vec![
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
]
);
}
#[test]
fn relay_operator_pubkeys_invalid_entry_is_error() {
let _guard = ENV_MUTEX.lock().unwrap();
std::env::set_var("RELAY_OPERATOR_PUBKEYS", "not-a-pubkey");
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_PUBKEYS")
));
}
#[test]
fn huddle_audio_available_can_be_disabled_for_horizontal_scaling() {
let _guard = ENV_MUTEX.lock().unwrap();

View File

@@ -0,0 +1,375 @@
//! Relay-operator community provisioning HTTP handler support.
//!
//! ## Authorization: operator, not owner
//!
//! Every other admin surface in this relay is community-scoped — the sender's
//! role is looked up in `relay_members (community_id, pubkey)` for the
//! host-resolved tenant. Community *creation* cannot work that way: its effect
//! is the creation of tenancy itself, so the authorizing identity must sit
//! above tenants. The gate here is the deployment-level
//! `RELAY_OPERATOR_PUBKEYS` allowlist (see `Config::relay_operator_pubkeys`).
//! An empty allowlist (the default) disables provisioning entirely.
//!
//! The public surface is `POST /operator/communities`, authenticated by NIP-98
//! and gated by the deployment-level `RELAY_OPERATOR_PUBKEYS` allowlist. The
//! endpoint is intentionally outside the Nostr event ingest data plane: no
//! relay-membership bypass, no special event kind, no storage or fan-out.
//!
//! ## Request shape
//!
//! ```json
//! { "host": "acme.communities.buzz.xyz", "initial_owner_pubkey": "<hex>" }
//! ```
//!
//! `initial_owner_pubkey` is optional. When present for an existing community,
//! it rotates that community owner through the same bootstrap path used by
//! `RELAY_OWNER_PUBKEY`; relay operators are deployment-root authorities.
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use tracing::info;
use buzz_core::tenant::normalize_host;
use url::{Host, Url};
use crate::state::AppState;
/// Maximum accepted authority length. Matches `communities.host VARCHAR(255)`.
const MAX_HOST_LEN: usize = 255;
/// JSON body for `POST /operator/communities`.
#[derive(Debug, Deserialize)]
pub struct ProvisionCommunityRequest {
/// Normalized authority for the community to ensure/create.
pub host: String,
/// Optional initial owner pubkey. When set on an existing community this
/// rotates the owner through the same bootstrap path used at startup.
#[serde(default)]
pub initial_owner_pubkey: Option<String>,
}
/// JSON response from `POST /operator/communities`.
#[derive(Debug, Serialize)]
pub struct ProvisionCommunityResponse {
/// UUID of the ensured/created community.
pub community_id: String,
/// Canonical host stored on the community row.
pub host: String,
/// `created` when the host row was inserted, `existed` when it was already
/// present and the request converged idempotently.
pub status: &'static str,
/// Echoes the validated owner pubkey when an owner bootstrap/rotation ran.
#[serde(skip_serializing_if = "Option::is_none")]
pub owner_pubkey: Option<String>,
}
pub(crate) fn validate_pubkey_hex(value: &str) -> Option<String> {
let normalized = value.to_ascii_lowercase();
(normalized.len() == 64 && normalized.chars().all(|c| c.is_ascii_hexdigit()))
.then_some(normalized)
}
/// Validate a normalized host authority value for a community.
///
/// The host must already be in normalized shape (`normalize_host` is a
/// no-op on it): lowercase, no default port, no trailing dot. Requiring the
/// caller to send the normalized form keeps the stored `communities.host`
/// value byte-identical to what request-time host resolution will look up.
fn validate_host(host: &str) -> Result<(), String> {
if host.is_empty() {
return Err("host is empty".to_string());
}
if host.len() > MAX_HOST_LEN {
return Err(format!(
"host too long: {} bytes (max {MAX_HOST_LEN})",
host.len()
));
}
if normalize_host(host) != host {
return Err(format!(
"host is not normalized: expected {:?}",
normalize_host(host)
));
}
validate_authority(host)
}
fn validate_authority(authority: &str) -> Result<(), String> {
if authority
.chars()
.any(|c| c.is_control() || c.is_whitespace())
{
return Err("host contains invalid characters".to_string());
}
if authority.contains('/')
|| authority.contains('?')
|| authority.contains('#')
|| authority.contains('@')
{
return Err(
"host must be a bare authority (no scheme, path, query, or userinfo)".to_string(),
);
}
// Parse as an HTTP authority by wrapping it in a URL. This rejects empty
// hosts, malformed bracketed IPv6 literals, and invalid ports while keeping
// the accepted shape aligned with request `Host` authority syntax.
let parsed = Url::parse(&format!("http://{authority}/"))
.map_err(|_| "host is not a valid authority".to_string())?;
let host = parsed
.host()
.ok_or_else(|| "host is not a valid authority".to_string())?;
let serialized_host = match host {
Host::Domain(domain) => {
validate_domain_labels(domain)?;
domain.to_string()
}
Host::Ipv4(addr) => addr.to_string(),
Host::Ipv6(addr) => format!("[{addr}]"),
};
let canonical_authority = match parsed.port() {
Some(port) => format!("{serialized_host}:{port}"),
None => serialized_host,
};
if canonical_authority != authority {
return Err(format!(
"host is not a canonical authority: expected {canonical_authority:?}"
));
}
Ok(())
}
fn validate_domain_labels(domain: &str) -> Result<(), String> {
if domain.len() > 253 {
return Err("domain name too long".to_string());
}
for label in domain.split('.') {
if label.is_empty() {
return Err("domain contains an empty label".to_string());
}
if label.len() > 63 {
return Err("domain label too long".to_string());
}
let valid_label = label
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'-')
&& !label.starts_with('-')
&& !label.ends_with('-');
if !valid_label {
return Err("domain label contains invalid characters".to_string());
}
}
Ok(())
}
/// Normalize and validate a host supplied to read-only operator endpoints.
///
/// Unlike create, availability checks may accept non-canonical but normalizable
/// authority values (uppercase host, trailing dot, default port) so kgoose can
/// ask the relay for the canonical spelling before creating. Schemes, paths,
/// userinfo, whitespace/control characters, and oversized values are still
/// rejected.
pub(crate) fn normalize_candidate_host(host: &str) -> Result<String, String> {
if host.is_empty() {
return Err("host is empty".to_string());
}
if host.len() > MAX_HOST_LEN {
return Err(format!(
"host too long: {} bytes (max {MAX_HOST_LEN})",
host.len()
));
}
if host.chars().any(|c| c.is_control() || c.is_whitespace()) {
return Err("host contains invalid characters".to_string());
}
if host.contains('/') || host.contains('?') || host.contains('#') || host.contains('@') {
return Err(
"host must be a bare authority (no scheme, path, query, or userinfo)".to_string(),
);
}
let normalized = normalize_host(host);
validate_host(&normalized)?;
Ok(normalized)
}
/// Validate and execute a relay-operator community provisioning request.
///
/// The caller is an HTTP operator endpoint, not the Nostr event ingest path.
/// That keeps the tenant data-plane fences unchanged: no relay-membership
/// bypass, no special event kind, no command routed ahead of moderation/write
/// blocks. The endpoint authenticates its NIP-98 signer first, then passes the
/// signer here for the deployment-level `RELAY_OPERATOR_PUBKEYS` allowlist.
///
/// Idempotency and owner semantics: the request is idempotent on the host row
/// (re-sending it never duplicates a community). When `initial_owner_pubkey` is
/// present, the owner is (re)bootstrapped via [`buzz_db::Db::bootstrap_owner`]
/// even if the community already existed — any previous owner is demoted to
/// admin, exactly like rotating `RELAY_OWNER_PUBKEY` for the deployment
/// community. This makes a retry after a partial failure (row created, owner
/// bootstrap crashed) converge, at the cost that an operator-signed request can
/// rotate an existing community's owner. The operator allowlist is therefore
/// documented as deployment-root authority, not create-only authority.
pub async fn provision_community(
state: &Arc<AppState>,
operator_pubkey: &nostr::PublicKey,
request: ProvisionCommunityRequest,
) -> Result<ProvisionCommunityResponse, String> {
let operator_hex = operator_pubkey.to_hex();
// Operator gate. Deliberately NOT a relay_members lookup: provisioning
// authority spans tenants and lives in deployment config only. Empty
// allowlist → everyone is rejected (fail closed).
if !state
.config
.relay_operator_pubkeys
.iter()
.any(|pk| pk == &operator_hex)
{
return Err("actor not authorized: not a relay operator".to_string());
}
validate_host(&request.host)?;
let initial_owner = request
.initial_owner_pubkey
.as_deref()
.map(|value| {
validate_pubkey_hex(value).ok_or_else(|| {
"invalid initial_owner_pubkey: expected 64-char hex pubkey".to_string()
})
})
.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.
let record = state
.db
.ensure_configured_community(&request.host)
.await
.map_err(|e| format!("failed to create community: {e}"))?;
if let Some(owner_hex) = &initial_owner {
state
.db
.bootstrap_owner(record.id, owner_hex)
.await
.map_err(|e| format!("community provisioned but owner bootstrap failed: {e}"))?;
}
info!(
operator = %operator_hex,
community = %record.id,
host = %record.host,
owner = initial_owner.as_deref().unwrap_or("<none>"),
created = record.created,
"community provisioned via operator endpoint"
);
Ok(ProvisionCommunityResponse {
community_id: record.id.to_string(),
host: record.host,
status: if record.created { "created" } else { "existed" },
owner_pubkey: initial_owner,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn host_valid_bare_domain() {
assert!(validate_host("acme.communities.buzz.xyz").is_ok());
}
#[test]
fn host_valid_with_port() {
assert!(validate_host("localhost:3000").is_ok());
}
#[test]
fn host_rejects_empty() {
assert!(validate_host("").is_err());
}
#[test]
fn host_rejects_uppercase() {
assert!(validate_host("Acme.example").is_err());
}
#[test]
fn host_rejects_default_port() {
assert!(validate_host("acme.example:443").is_err());
assert!(validate_host("acme.example:80").is_err());
}
#[test]
fn host_rejects_trailing_dot() {
assert!(validate_host("acme.example.").is_err());
}
#[test]
fn host_rejects_scheme_path_userinfo() {
assert!(validate_host("wss://acme.example").is_err());
assert!(validate_host("acme.example/path").is_err());
assert!(validate_host("user@acme.example").is_err());
assert!(validate_host("acme.example?x=1").is_err());
assert!(validate_host("acme.example#frag").is_err());
}
#[test]
fn host_rejects_invalid_authorities() {
assert!(validate_host(":").is_err());
assert!(validate_host("example..com").is_err());
assert!(validate_host("foo_bar.example").is_err());
assert!(validate_host("-bad.example").is_err());
assert!(validate_host("bad-.example").is_err());
assert!(validate_host("example.com:99999").is_err());
assert!(validate_host("[::1").is_err());
assert!(validate_host("[not-ipv6]").is_err());
assert!(validate_host(&format!("{}.example", "a".repeat(64))).is_err());
}
#[test]
fn host_rejects_whitespace_and_control() {
assert!(validate_host("acme .example").is_err());
assert!(validate_host("acme\n.example").is_err());
}
#[test]
fn host_rejects_oversized() {
let long = format!("{}.example", "a".repeat(260));
assert!(validate_host(&long).is_err());
}
#[test]
fn host_accepts_ipv6_bracket_literal() {
assert!(validate_host("[::1]:3000").is_ok());
}
#[test]
fn candidate_host_normalizes_safe_variants() {
assert_eq!(
normalize_candidate_host("Acme.Example:443").unwrap(),
"acme.example"
);
assert_eq!(
normalize_candidate_host("acme.example.").unwrap(),
"acme.example"
);
}
#[test]
fn candidate_host_rejects_non_authorities() {
assert!(normalize_candidate_host("https://acme.example").is_err());
assert!(normalize_candidate_host("acme.example/path").is_err());
assert!(normalize_candidate_host("acme .example").is_err());
}
}

View File

@@ -200,7 +200,7 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result<Scope, &'static s
Ok(Scope::AdminChannels)
}
// NIP-43: relay membership admin commands (9030–9032) + Buzz
// workspace-profile command (9033)
// workspace-profile command (9033).
k if k == RELAY_ADMIN_ADD_MEMBER
|| k == RELAY_ADMIN_REMOVE_MEMBER
|| k == RELAY_ADMIN_CHANGE_ROLE

View File

@@ -4,6 +4,8 @@ pub mod auth;
pub mod close;
/// Command executor — transactional processing for command kinds.
pub mod command_executor;
/// Relay-operator community provisioning HTTP support.
pub mod community_provisioning;
/// NIP-45 COUNT handler.
pub mod count;
/// EVENT handler — WS dispatcher → ingest pipeline → fan-out.

View File

@@ -60,6 +60,14 @@ pub fn build_router(state: Arc<AppState>) -> Router {
.route("/events", post(api::bridge::submit_event))
.route("/query", post(api::bridge::query_events))
.route("/count", post(api::bridge::count_events))
.route(
"/operator/communities",
get(api::operator::list_owned_communities).post(api::operator::provision_community),
)
.route(
"/operator/communities/availability",
get(api::operator::community_availability),
)
// Moderation queue reads (NIP-98 auth + mod-authz gate, L6)
.route("/moderation/reports", get(api::bridge::moderation_reports))
.route("/moderation/audit", get(api::bridge::moderation_audit))
@@ -96,6 +104,7 @@ pub fn build_router(state: Arc<AppState>) -> Router {
// Reserved API prefixes must 404 normally, not serve index.html.
let reserved = path.starts_with("/api/")
|| path.starts_with("/media/")
|| path.starts_with("/operator/")
|| path.starts_with("/git/")
|| path.starts_with("/internal/")
|| path.starts_with("/.well-known/")