jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-01 15:01:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(relay): add corporate identity binding

Browse Source 
Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com>
This commit is contained in:
Franco Sola
2026-07-02 14:20:01 -06:00
committed by Cea Stapleton Cordasco
parent 2c0ac24674
commit b53bcfdd6d
22 changed files with 1807 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
.env.example
View File

@@ -69,6 +69,32 @@ RELAY_URL=ws://localhost:3000
# BUZZ_RATE_LIMIT_AGENT_ELEVATED_MESSAGES_PER_MIN=300
# BUZZ_RATE_LIMIT_AGENT_PLATFORM_MESSAGES_PER_MIN=600
# Corporate identity verification (disabled by default). When enabled, the relay
# requires authenticated requests to present a valid corporate JWT, then binds
# the configured uid claim to the Nostr pubkey proven by NIP-42/NIP-98. The JWT
# may be injected by a trusted proxy or attached by a first-party client; the
# relay treats both as the same header. Client token forwarding and admin
# revocation/rotation workflows are separate follow-up features.
#
# Operational notes for the initial implementation:
# - When a trusted proxy injects this header, it MUST overwrite any inbound
# client-supplied value before forwarding to the relay.
# - A uid/pubkey conflict is a hard lockout until an operator resolves the
# binding out of band; admin revocation/rotation endpoints are not in this PR.
# - JWKS outages fail closed for human JWT authentication. Delegated agent
# admission can still work when the owner binding is already present.
# - DISPLAY_CLAIM is stored in identity_bindings.display_name. Using email here
# stores corporate PII and PR2 may surface it in clients.
# BUZZ_REQUIRE_CORPORATE_IDENTITY=false
# BUZZ_CORPORATE_IDENTITY_JWT_HEADER=x-forwarded-identity-token
# BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION=true
# BUZZ_CORPORATE_IDENTITY_JWKS_URI=https://idp.example/.well-known/jwks.json
# BUZZ_CORPORATE_IDENTITY_ISSUER=https://idp.example
# BUZZ_CORPORATE_IDENTITY_AUDIENCE=buzz-relay
# BUZZ_CORPORATE_IDENTITY_UID_CLAIM=sub
# BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM=email
# BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM=buzz_npub
# -----------------------------------------------------------------------------
# Git (NIP-34 bare repositories)
# -----------------------------------------------------------------------------

28
Cargo.lock generated
View File

@@ -1207,6 +1207,7 @@ dependencies = [
"hex",
"hmac 0.13.0",
"infer",
"jsonwebtoken",
"mesh-llm-host-runtime",
"mesh-llm-sdk",
"metrics",
@@ -4349,6 +4350,21 @@ dependencies = [
"serde_json",
]
[[package]]
name = "jsonwebtoken"
version = "9.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde"
dependencies = [
"base64 0.22.1",
"js-sys",
"pem",
"ring",
"serde",
"serde_json",
"simple_asn1",
]
[[package]]
name = "kasuari"
version = "0.4.12"
@@ -8726,6 +8742,18 @@ dependencies = [
"bitflags 2.13.0",
]
[[package]]
name = "simple_asn1"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d"
dependencies = [
"num-bigint",
"num-traits",
"thiserror 2.0.18",
"time",
]
[[package]]
name = "siphasher"
version = "1.0.3"

1
Cargo.toml
View File

@@ -77,6 +77,7 @@ postcard = { version = "1", default-features = false, features = ["use-std"]
iroh = { version = "1.0.0-rc.0", default-features = false, features = ["tls-ring"] }
serde_json = "1"
serde_yaml = "0.9"
jsonwebtoken = "9"
evalexpr = "11"
cron = "0.16"
# Observability

8
crates/buzz-audit/src/action.rs
View File

@@ -28,6 +28,10 @@ pub enum AuditAction {
RateLimitExceeded,
/// A media file was uploaded via the Blossom endpoint.
MediaUploaded,
/// A corporate identity binding was created.
CorporateIdentityBindingCreated,
/// A corporate identity binding attempt conflicted with an active binding.
CorporateIdentityBindingConflict,
}
impl AuditAction {
@@ -45,6 +49,8 @@ impl AuditAction {
Self::AuthFailure => "auth_failure",
Self::RateLimitExceeded => "rate_limit_exceeded",
Self::MediaUploaded => "media_uploaded",
Self::CorporateIdentityBindingCreated => "corporate_identity_binding_created",
Self::CorporateIdentityBindingConflict => "corporate_identity_binding_conflict",
}
}
@@ -60,6 +66,8 @@ impl AuditAction {
Self::AuthFailure,
Self::RateLimitExceeded,
Self::MediaUploaded,
Self::CorporateIdentityBindingCreated,
Self::CorporateIdentityBindingConflict,
];
}

457
crates/buzz-db/src/identity_binding.rs Normal file
View File

@@ -0,0 +1,457 @@
//! Corporate identity binding persistence.
//!
//! Bindings map a corporate IdP uid to the currently authorized Nostr pubkey
//! inside one Buzz community. The active uniqueness indexes deliberately model
//! one active pubkey per uid and one active uid per pubkey. Rotation/revocation
//! flows clear that active state in a follow-up lifecycle layer rather than
//! silently rewriting it during authentication.
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Postgres, Row, Transaction};
use crate::error::{DbError, Result};
use buzz_core::CommunityId;
/// Binding source when the IdP JWT carries the pubkey claim.
pub const SOURCE_JWT_NPUB: &str = "jwt_npub";
/// Binding source when the relay falls back to the stored uid/pubkey binding.
pub const SOURCE_DB_BINDING: &str = "db_binding";
/// Active corporate identity binding row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IdentityBinding {
/// Corporate IdP subject or configured stable uid claim.
pub uid: String,
/// Bound Nostr pubkey bytes.
pub pubkey: Vec<u8>,
/// Human-readable display claim captured from the latest accepted JWT.
pub display_name: Option<String>,
/// Source that established or last strengthened the active binding.
pub source: String,
/// When the binding was first created.
pub created_at: DateTime<Utc>,
/// When the binding row was last updated.
pub updated_at: DateTime<Utc>,
/// When the binding was last seen during authentication.
pub last_seen_at: DateTime<Utc>,
}
/// Existing active binding that conflicts with a requested binding.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IdentityBindingConflict {
/// Existing active uid.
pub uid: String,
/// Existing active pubkey bytes.
pub pubkey: Vec<u8>,
/// Existing active binding source.
pub source: String,
}
/// Outcome of creating or validating a corporate identity binding.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BindIdentityResult {
/// A new active binding was created.
Created,
/// The requested binding matched an existing active binding.
Matched,
/// Another active binding already owns the uid or pubkey.
Conflict(IdentityBindingConflict),
}
fn validate_inputs(uid: &str, pubkey: &[u8], source: &str) -> Result<()> {
if uid.trim().is_empty() {
return Err(DbError::InvalidData(
"identity binding uid must not be empty".to_string(),
));
}
validate_pubkey(pubkey)?;
if !matches!(source, SOURCE_JWT_NPUB | SOURCE_DB_BINDING) {
return Err(DbError::InvalidData(format!(
"invalid identity binding source: {source}"
)));
}
Ok(())
}
fn validate_pubkey(pubkey: &[u8]) -> Result<()> {
if pubkey.len() != 32 {
return Err(DbError::InvalidData(
"identity binding pubkey must be 32 bytes".to_string(),
));
}
Ok(())
}
fn row_to_binding(row: sqlx::postgres::PgRow) -> Result<IdentityBinding> {
Ok(IdentityBinding {
uid: row.try_get("uid")?,
pubkey: row.try_get("pubkey")?,
display_name: row.try_get("display_name")?,
source: row.try_get("source")?,
created_at: row.try_get("created_at")?,
updated_at: row.try_get("updated_at")?,
last_seen_at: row.try_get("last_seen_at")?,
})
}
async fn active_by_uid_tx(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
uid: &str,
) -> Result<Option<IdentityBinding>> {
let row = sqlx::query(
r#"
SELECT uid, pubkey, display_name, source, created_at, updated_at, last_seen_at
FROM identity_bindings
WHERE community_id = $1 AND uid = $2 AND revoked_at IS NULL
FOR UPDATE
"#,
)
.bind(community_id.as_uuid())
.bind(uid)
.fetch_optional(&mut **tx)
.await?;
row.map(row_to_binding).transpose()
}
async fn active_by_pubkey_tx(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<IdentityBinding>> {
let row = sqlx::query(
r#"
SELECT uid, pubkey, display_name, source, created_at, updated_at, last_seen_at
FROM identity_bindings
WHERE community_id = $1 AND pubkey = $2 AND revoked_at IS NULL
FOR UPDATE
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_optional(&mut **tx)
.await?;
row.map(row_to_binding).transpose()
}
fn conflict_from(binding: IdentityBinding) -> IdentityBindingConflict {
IdentityBindingConflict {
uid: binding.uid,
pubkey: binding.pubkey,
source: binding.source,
}
}
async fn lock_identity_keys_tx(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
uid: &str,
pubkey: &[u8],
) -> Result<()> {
let mut keys = [
format!("{}:uid:{uid}", community_id.as_uuid()),
format!("{}:pubkey:{}", community_id.as_uuid(), hex::encode(pubkey)),
];
keys.sort();
for key in keys {
sqlx::query("SELECT pg_advisory_xact_lock(hashtext('identity_bindings'), hashtext($1))")
.bind(key)
.execute(&mut **tx)
.await?;
}
Ok(())
}
/// Create or validate an active corporate identity binding.
///
/// This is a fail-closed auth-time operation:
/// - same uid + same pubkey updates display/last_seen and succeeds;
/// - same uid + different pubkey conflicts;
/// - same pubkey + different uid conflicts;
/// - no active row creates a new binding.
pub async fn bind_or_validate_identity(
pool: &PgPool,
community_id: CommunityId,
uid: &str,
pubkey: &[u8],
display_name: Option<&str>,
source: &str,
) -> Result<BindIdentityResult> {
validate_inputs(uid, pubkey, source)?;
let mut tx = pool.begin().await?;
sqlx::query("SET LOCAL lock_timeout = '3s'")
.execute(&mut *tx)
.await?;
lock_identity_keys_tx(&mut tx, community_id, uid, pubkey).await?;
let active_uid = active_by_uid_tx(&mut tx, community_id, uid).await?;
if let Some(binding) = active_uid {
if binding.pubkey != pubkey {
tx.rollback().await?;
return Ok(BindIdentityResult::Conflict(conflict_from(binding)));
}
sqlx::query(
r#"
UPDATE identity_bindings
SET display_name = $4,
source = CASE
WHEN source = 'jwt_npub' AND $5 = 'db_binding' THEN source
ELSE $5
END,
updated_at = NOW(),
last_seen_at = NOW()
WHERE community_id = $1 AND uid = $2 AND pubkey = $3 AND revoked_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(uid)
.bind(pubkey)
.bind(display_name)
.bind(source)
.execute(&mut *tx)
.await?;
tx.commit().await?;
return Ok(BindIdentityResult::Matched);
}
let active_pubkey = active_by_pubkey_tx(&mut tx, community_id, pubkey).await?;
if let Some(binding) = active_pubkey {
if binding.uid != uid {
tx.rollback().await?;
return Ok(BindIdentityResult::Conflict(conflict_from(binding)));
}
}
sqlx::query(
r#"
INSERT INTO identity_bindings (community_id, uid, pubkey, display_name, source)
VALUES ($1, $2, $3, $4, $5)
"#,
)
.bind(community_id.as_uuid())
.bind(uid)
.bind(pubkey)
.bind(display_name)
.bind(source)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(BindIdentityResult::Created)
}
/// Return the active binding for `pubkey`, if one exists.
pub async fn get_active_identity_binding_by_pubkey(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<IdentityBinding>> {
validate_pubkey(pubkey)?;
let row = sqlx::query(
r#"
SELECT uid, pubkey, display_name, source, created_at, updated_at, last_seen_at
FROM identity_bindings
WHERE community_id = $1 AND pubkey = $2 AND revoked_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_optional(pool)
.await?;
row.map(row_to_binding).transpose()
}
#[cfg(test)]
mod tests {
use super::*;
use nostr::Keys;
use uuid::Uuid;
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
async fn setup_pool() -> PgPool {
let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
.unwrap_or_else(|_| TEST_DB_URL.to_owned());
let pool = PgPool::connect(&database_url)
.await
.expect("connect to test DB");
crate::migration::run_migrations(&pool)
.await
.expect("run migrations");
pool
}
async fn make_community(pool: &PgPool) -> CommunityId {
let id = Uuid::new_v4();
let host = format!("identity-binding-test-{}.example", id.simple());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(host)
.execute(pool)
.await
.expect("insert test community");
CommunityId::from_uuid(id)
}
fn random_pubkey() -> Vec<u8> {
Keys::generate().public_key().to_bytes().to_vec()
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn bind_identity_creates_then_matches_idempotently() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let pubkey = random_pubkey();
let created = bind_or_validate_identity(
&pool,
community,
"user-1",
&pubkey,
Some("first@example.com"),
SOURCE_DB_BINDING,
)
.await
.expect("create binding");
assert_eq!(created, BindIdentityResult::Created);
let matched = bind_or_validate_identity(
&pool,
community,
"user-1",
&pubkey,
Some("second@example.com"),
SOURCE_JWT_NPUB,
)
.await
.expect("match existing binding");
assert_eq!(matched, BindIdentityResult::Matched);
let binding = get_active_identity_binding_by_pubkey(&pool, community, &pubkey)
.await
.expect("lookup binding")
.expect("binding exists");
assert_eq!(binding.uid, "user-1");
assert_eq!(binding.display_name.as_deref(), Some("second@example.com"));
assert_eq!(binding.source, SOURCE_JWT_NPUB);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn bind_identity_rejects_uid_conflict() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let original_pubkey = random_pubkey();
let conflicting_pubkey = random_pubkey();
bind_or_validate_identity(
&pool,
community,
"user-1",
&original_pubkey,
Some("user@example.com"),
SOURCE_DB_BINDING,
)
.await
.expect("create binding");
let result = bind_or_validate_identity(
&pool,
community,
"user-1",
&conflicting_pubkey,
Some("user@example.com"),
SOURCE_DB_BINDING,
)
.await
.expect("uid conflict is a binding result");
assert_eq!(
result,
BindIdentityResult::Conflict(IdentityBindingConflict {
uid: "user-1".to_string(),
pubkey: original_pubkey,
source: SOURCE_DB_BINDING.to_string(),
})
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn bind_identity_rejects_pubkey_conflict() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let pubkey = random_pubkey();
bind_or_validate_identity(
&pool,
community,
"user-1",
&pubkey,
Some("user@example.com"),
SOURCE_DB_BINDING,
)
.await
.expect("create binding");
let result = bind_or_validate_identity(
&pool,
community,
"user-2",
&pubkey,
Some("other@example.com"),
SOURCE_JWT_NPUB,
)
.await
.expect("pubkey conflict is a binding result");
assert_eq!(
result,
BindIdentityResult::Conflict(IdentityBindingConflict {
uid: "user-1".to_string(),
pubkey,
source: SOURCE_DB_BINDING.to_string(),
})
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn bind_identity_does_not_downgrade_jwt_npub_source() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let pubkey = random_pubkey();
bind_or_validate_identity(
&pool,
community,
"user-1",
&pubkey,
Some("user@example.com"),
SOURCE_JWT_NPUB,
)
.await
.expect("create strong binding");
let matched = bind_or_validate_identity(
&pool,
community,
"user-1",
&pubkey,
Some("user@example.com"),
SOURCE_DB_BINDING,
)
.await
.expect("match existing binding");
assert_eq!(matched, BindIdentityResult::Matched);
let binding = get_active_identity_binding_by_pubkey(&pool, community, &pubkey)
.await
.expect("lookup binding")
.expect("binding exists");
assert_eq!(binding.source, SOURCE_JWT_NPUB);
}
}

32
crates/buzz-db/src/lib.rs
View File

@@ -27,6 +27,8 @@ pub mod event;
pub mod feed;
/// Git repository name registry (NIP-34 kind:30617).
pub mod git_repo;
/// Corporate identity binding persistence.
pub mod identity_binding;
/// Embedded database migrations.
pub mod migration;
/// Community moderation: reports, bans/timeouts, audit actions.
@@ -2535,6 +2537,36 @@ impl Db {
user::search_users(&self.pool, community_id, query, limit).await
}
/// Create or validate a corporate identity binding.
pub async fn bind_or_validate_identity(
&self,
community_id: CommunityId,
uid: &str,
pubkey: &[u8],
display_name: Option<&str>,
source: &str,
) -> Result<identity_binding::BindIdentityResult> {
identity_binding::bind_or_validate_identity(
&self.pool,
community_id,
uid,
pubkey,
display_name,
source,
)
.await
}
/// Return the active corporate identity binding for `pubkey`, if any.
pub async fn get_active_identity_binding_by_pubkey(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<identity_binding::IdentityBinding>> {
identity_binding::get_active_identity_binding_by_pubkey(&self.pool, community_id, pubkey)
.await
}
/// Atomically set agent owner — only if no owner is currently assigned.
/// Returns Ok(true) if set, Ok(false) if an owner already exists.
pub async fn set_agent_owner(

15
crates/buzz-db/src/migration.rs
View File

@@ -596,7 +596,6 @@ mod tests {
.as_str()
.contains("CREATE TABLE git_repo_names"));
assert!(!migrations[0].sql.as_str().contains("git_repo_names"));
// Same additive-migration rule for the per-community workspace icon
// (NIP-11 `icon`): its own version, never folded into 0001.
assert_eq!(migrations[2].version, 3);
@@ -880,7 +879,6 @@ mod tests {
.to_lowercase()
.contains("for update"));
assert!(ttl_shared.contains("NEW.kind <> 9007"));
// Use-limited invite links: durable relay_invites table stores only
// the SHA-256 of an opaque v2 code, scoped by community_id. Never
// listed in _operator_global_tables — it is community-scoped.
@@ -905,7 +903,6 @@ mod tests {
desired_schema.contains("CREATE TABLE join_policy_acceptances"),
"desired-state schema must include join-policy evidence used by invite claims",
);
// Replica heartbeat (this branch, renumbered to 0026 after
// 0025_relay_invites landed on main): the fence's portable read-side
// observation. A single CHECK'd row makes the token update the
@@ -919,6 +916,17 @@ mod tests {
assert!(heartbeat.contains("epoch"));
assert!(heartbeat.contains("INSERT INTO replica_heartbeat (id) VALUES (1)"));
assert!(heartbeat.contains("_operator_global_tables"));
// Corporate identity bindings are additive and community-scoped.
assert_eq!(migrations[26].version, 27);
assert!(migrations[26]
.sql
.as_str()
.contains("CREATE TABLE identity_bindings"));
assert!(migrations[26]
.sql
.as_str()
.contains("idx_identity_bindings_active_uid"));
}
#[test]
@@ -1247,6 +1255,7 @@ mod tests {
"communities",
"events",
"channels",
"identity_bindings",
"scheduled_workflow_fires",
"audit_log",
] {

1
crates/buzz-relay/Cargo.toml
View File

@@ -37,6 +37,7 @@ tower-http = { workspace = true }
nostr = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
jsonwebtoken = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
tracing-opentelemetry = { workspace = true }

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

@@ -175,6 +175,29 @@ async fn check_nip98_replay_with_guard(
}
}
async fn enforce_bridge_corporate_identity(
state: &AppState,
tenant: &TenantContext,
headers: &HeaderMap,
pubkey: nostr::PublicKey,
auth_tag: Option<&str>,
) -> Result<(), (StatusCode, Json<Value>)> {
let identity_jwt = crate::corporate_identity::identity_jwt_from_headers(
headers,
&state.config.corporate_identity,
);
crate::corporate_identity::enforce_corporate_identity(
state,
tenant.community(),
pubkey,
identity_jwt.as_deref(),
auth_tag,
)
.await
.map(|_| ())
.map_err(|e| e.into_api_error())
}
/// Construct the NIP-98 `u`-tag expected URL for a request bound to `tenant`.
///
/// Conformance row 44 obligation: "NIP-98 `u` URL host must match
@@ -801,6 +824,14 @@ async fn submit_event_authed(
// 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());
if let Err(e) =
enforce_bridge_corporate_identity(state, tenant, headers, pubkey, auth_tag).await
{
return SubmitOutcome::Err {
status: e.0,
response: e,
};
}
let nip_oa_owner = match super::relay_members::enforce_relay_membership(
state,
tenant.community(),
@@ -961,6 +992,7 @@ async fn query_events_authed(
let pubkey_bytes = pubkey.to_bytes().to_vec();
let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok());
enforce_bridge_corporate_identity(state, tenant, headers, pubkey, auth_tag).await?;
super::relay_members::enforce_relay_membership(
state,
tenant.community(),
@@ -1402,6 +1434,7 @@ async fn count_events_authed(
let pubkey_bytes = pubkey.to_bytes().to_vec();
let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok());
enforce_bridge_corporate_identity(state, tenant, headers, pubkey, auth_tag).await?;
super::relay_members::enforce_relay_membership(
state,
tenant.community(),

16
crates/buzz-relay/src/api/git/transport.rs
View File

@@ -211,6 +211,22 @@ impl axum::extract::FromRequestParts<Arc<AppState>> for GitAuth {
.get("x-auth-tag")
.and_then(|value| value.to_str().ok());
let auth_tag = event_auth_tag.as_deref().or(header_auth_tag);
let identity_jwt = crate::corporate_identity::identity_jwt_from_headers(
&parts.headers,
&state.config.corporate_identity,
);
if let Err(e) = crate::corporate_identity::enforce_corporate_identity(
state,
tenant.community(),
pubkey,
identity_jwt.as_deref(),
auth_tag,
)
.await
{
warn!(pubkey = %pubkey.to_hex(), error = %e, "git: corporate identity denied");
return Err((e.status_code(), e.public_message()).into_response());
}
if crate::api::relay_members::enforce_relay_membership(
state,
tenant.community(),

21
crates/buzz-relay/src/api/media.rs
View File

@@ -208,6 +208,27 @@ impl FromRequestParts<Arc<AppState>> for AuthenticatedUpload {
// media). On open relays (membership disabled) any valid Blossom signer
// may upload, matching the WS door's admission policy.
let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok());
let identity_jwt = crate::corporate_identity::identity_jwt_from_headers(
headers,
&state.config.corporate_identity,
);
crate::corporate_identity::enforce_corporate_identity(
state,
tenant.community(),
auth_event.pubkey,
identity_jwt.as_deref(),
auth_tag,
)
.await
.map_err(|e| {
tracing::warn!(pubkey = %auth_event.pubkey.to_hex(), error = %e, "media: corporate identity denied");
if e.status_code() == StatusCode::UNAUTHORIZED {
MediaError::Unauthorized
} else {
MediaError::RelayMembershipRequired
}
})?;
crate::api::relay_members::enforce_relay_membership(
state,
tenant.community(),

47
crates/buzz-relay/src/audio/handler.rs
View File

@@ -86,7 +86,6 @@ pub async fn ws_audio_handler(
.into_response();
}
};
let permit = match acquire_audio_connection_permit(&state.conn_semaphore) {
Some(permit) => permit,
None => {
@@ -98,12 +97,23 @@ pub async fn ws_audio_handler(
.into_response();
}
};
let corporate_identity_jwt = crate::corporate_identity::identity_jwt_from_headers(
&headers,
&state.config.corporate_identity,
);
// Keep the parser boundary at the largest message this route accepts. The
// checks in the receive loop still distinguish text from binary policy, but
// they run after tungstenite has assembled a message.
limit_audio_websocket(ws).on_upgrade(move |socket| {
handle_audio_connection(socket, state, tenant, channel_id, permit)
handle_audio_connection(
socket,
state,
tenant,
channel_id,
permit,
corporate_identity_jwt,
)
})
}
@@ -147,6 +157,7 @@ async fn handle_audio_connection(
tenant: TenantContext,
channel_id: Uuid,
_permit: OwnedSemaphorePermit,
corporate_identity_jwt: Option<String>,
) {
let cancel = CancellationToken::new();
let community_id = tenant.community();
@@ -159,7 +170,16 @@ async fn handle_audio_connection(
community_id,
cancel.clone(),
move || async move { check_state.db.is_community_active(community_id).await },
move || handle_active_audio_connection(socket, run_state, tenant, channel_id, cancel),
move || {
handle_active_audio_connection(
socket,
run_state,
tenant,
channel_id,
cancel,
corporate_identity_jwt,
)
},
)
.await;
}
@@ -170,6 +190,7 @@ async fn handle_active_audio_connection(
tenant: TenantContext,
channel_id: Uuid,
cancel: CancellationToken,
corporate_identity_jwt: Option<String>,
) {
let (mut ws_send, mut ws_recv) = socket.split();
@@ -241,6 +262,26 @@ async fn handle_active_audio_connection(
let pubkey_bytes = pubkey.to_bytes().to_vec();
let parent_channel_id = auth_msg.parent_channel_id;
if let Err(e) = crate::corporate_identity::enforce_corporate_identity(
&state,
tenant.community(),
pubkey,
corporate_identity_jwt.as_deref(),
auth_tag_json.as_deref(),
)
.await
{
warn!(channel_id = %channel_id, pubkey = %pubkey_hex, error = %e, "audio: corporate identity denied");
let _ = ws_send
.send(WsMessage::Text(
serde_json::json!({"type": "error", "message": e.public_message()})
.to_string()
.into(),
))
.await;
return;
}
if crate::api::relay_members::enforce_relay_membership(
&state,
tenant.community(),

198
crates/buzz-relay/src/config.rs
View File

@@ -13,6 +13,13 @@ use tracing::warn;
/// NIP-44 encryption overhead.
pub const DEFAULT_MAX_FRAME_BYTES: usize = 512 * 1024;
/// Default header carrying a corporate identity JWT.
pub const DEFAULT_CORPORATE_IDENTITY_JWT_HEADER: &str = "x-forwarded-identity-token";
/// Default JWT claim used as the stable corporate uid.
pub const DEFAULT_CORPORATE_IDENTITY_UID_CLAIM: &str = "sub";
/// Default JWT claim displayed as the verified corporate identity.
pub const DEFAULT_CORPORATE_IDENTITY_DISPLAY_CLAIM: &str = "email";
/// Errors that can occur while loading relay configuration.
#[derive(Debug, Error)]
pub enum ConfigError {
@@ -46,6 +53,50 @@ pub struct JoinPolicyConfig {
pub version: String,
}
/// Source-neutral corporate identity configuration.
///
/// The relay does not care whether the JWT was injected by a trusted proxy or
/// attached by a first-party client. It only validates the JWT and binds the
/// configured uid claim to the authenticated Nostr pubkey after NIP proof.
#[derive(Debug, Clone)]
pub struct CorporateIdentityConfig {
/// Whether every authenticated request must satisfy corporate identity.
pub require: bool,
/// Header containing the corporate identity JWT.
pub jwt_header: String,
/// Allow agents without JWTs to pass the corporate identity gate through
/// NIP-OA when their owner pubkey already has an active identity binding.
pub allow_delegation: bool,
/// JWKS URI used to verify JWT signatures.
pub jwks_uri: String,
/// Expected JWT issuer.
pub issuer: String,
/// Expected JWT audience.
pub audience: String,
/// Claim name used as Buzz's stable corporate uid.
pub uid_claim: String,
/// Claim name used for verified display.
pub display_claim: String,
/// Optional claim name carrying a hex pubkey or `npub1...`.
pub npub_claim: Option<String>,
}
impl Default for CorporateIdentityConfig {
fn default() -> Self {
Self {
require: false,
jwt_header: DEFAULT_CORPORATE_IDENTITY_JWT_HEADER.to_string(),
allow_delegation: true,
jwks_uri: String::new(),
issuer: String::new(),
audience: String::new(),
uid_claim: DEFAULT_CORPORATE_IDENTITY_UID_CLAIM.to_string(),
display_claim: DEFAULT_CORPORATE_IDENTITY_DISPLAY_CLAIM.to_string(),
npub_claim: None,
}
}
}
/// Relay runtime configuration, loaded from environment variables.
#[derive(Debug, Clone)]
pub struct Config {
@@ -200,6 +251,9 @@ pub struct Config {
/// Default: `false`. Set via `BUZZ_ALLOW_NIP_OA_AUTH=true`.
pub allow_nip_oa_auth: bool,
/// Corporate identity verification and uid/pubkey binding.
pub corporate_identity: CorporateIdentityConfig,
/// Media storage configuration (S3/MinIO).
pub media: buzz_media::MediaConfig,
/// Maximum concurrent media uploads handled by one relay process.
@@ -417,6 +471,77 @@ fn ensure_git_path(
Ok(git_repo_path)
}
fn env_bool(name: &str, default: bool) -> bool {
std::env::var(name)
.map(|v| {
matches!(
v.trim().to_ascii_lowercase().as_str(),
"true" | "1" | "yes" | "on"
)
})
.unwrap_or(default)
}
fn env_trimmed(name: &str) -> Option<String> {
std::env::var(name)
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
}
fn load_corporate_identity_config() -> Result<CorporateIdentityConfig, ConfigError> {
let mut config = CorporateIdentityConfig::default();
config.require = env_bool("BUZZ_REQUIRE_CORPORATE_IDENTITY", config.require);
config.jwt_header = env_trimmed("BUZZ_CORPORATE_IDENTITY_JWT_HEADER")
.unwrap_or_else(|| config.jwt_header.clone())
.to_ascii_lowercase();
config.allow_delegation = env_bool(
"BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION",
config.allow_delegation,
);
config.jwks_uri =
env_trimmed("BUZZ_CORPORATE_IDENTITY_JWKS_URI").unwrap_or_else(|| config.jwks_uri.clone());
config.issuer =
env_trimmed("BUZZ_CORPORATE_IDENTITY_ISSUER").unwrap_or_else(|| config.issuer.clone());
config.audience =
env_trimmed("BUZZ_CORPORATE_IDENTITY_AUDIENCE").unwrap_or_else(|| config.audience.clone());
config.uid_claim = env_trimmed("BUZZ_CORPORATE_IDENTITY_UID_CLAIM")
.unwrap_or_else(|| config.uid_claim.clone());
config.display_claim = env_trimmed("BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM")
.unwrap_or_else(|| config.display_claim.clone());
config.npub_claim = env_trimmed("BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM");
if config.require {
let mut missing = Vec::new();
if config.jwt_header.is_empty() {
missing.push("BUZZ_CORPORATE_IDENTITY_JWT_HEADER");
}
if config.jwks_uri.is_empty() {
missing.push("BUZZ_CORPORATE_IDENTITY_JWKS_URI");
}
if config.issuer.is_empty() {
missing.push("BUZZ_CORPORATE_IDENTITY_ISSUER");
}
if config.audience.is_empty() {
missing.push("BUZZ_CORPORATE_IDENTITY_AUDIENCE");
}
if config.uid_claim.is_empty() {
missing.push("BUZZ_CORPORATE_IDENTITY_UID_CLAIM");
}
if config.display_claim.is_empty() {
missing.push("BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM");
}
if !missing.is_empty() {
return Err(ConfigError::InvalidValue(format!(
"BUZZ_REQUIRE_CORPORATE_IDENTITY=true but required corporate identity config is missing: {}",
missing.join(", ")
)));
}
}
Ok(config)
}
impl Config {
/// Loads configuration from environment variables, falling back to development defaults.
pub fn from_env() -> Result<Self, ConfigError> {
@@ -570,6 +695,8 @@ impl Config {
.map(|v| v == "true" || v == "1")
.unwrap_or(false);
let corporate_identity = load_corporate_identity_config()?;
// Note: intentionally not prefixed with BUZZ_ — this is a relay-identity
// config that may be shared across multiple services (e.g., ACP agent).
let relay_owner_pubkey = std::env::var("RELAY_OWNER_PUBKEY")
@@ -961,6 +1088,7 @@ impl Config {
relay_operator_api_origin,
relay_operator_pubkeys,
allow_nip_oa_auth,
corporate_identity,
media,
media_max_concurrent_uploads,
media_max_concurrent_uploads_per_pubkey,
@@ -997,9 +1125,26 @@ mod tests {
// value set by `invalid_bind_addr_returns_error`, causing a flaky failure.
static ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn clear_corporate_identity_env() {
for name in [
"BUZZ_REQUIRE_CORPORATE_IDENTITY",
"BUZZ_CORPORATE_IDENTITY_JWT_HEADER",
"BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION",
"BUZZ_CORPORATE_IDENTITY_JWKS_URI",
"BUZZ_CORPORATE_IDENTITY_ISSUER",
"BUZZ_CORPORATE_IDENTITY_AUDIENCE",
"BUZZ_CORPORATE_IDENTITY_UID_CLAIM",
"BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM",
"BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM",
] {
std::env::remove_var(name);
}
}
#[test]
fn defaults_are_valid() {
let _guard = ENV_MUTEX.lock().unwrap();
clear_corporate_identity_env();
let config = Config::from_env().expect("default config");
assert!(config.bind_addr.port() > 0);
assert!(!config.database_url.is_empty());
@@ -1051,6 +1196,59 @@ mod tests {
config.huddle_audio_available,
"huddle_audio_available should default to true so single-pod (N=1) keeps today's huddle behavior"
);
assert!(
!config.corporate_identity.require,
"corporate identity should default to disabled"
);
assert_eq!(
config.corporate_identity.jwt_header,
DEFAULT_CORPORATE_IDENTITY_JWT_HEADER
);
assert!(
config.corporate_identity.allow_delegation,
"corporate identity delegation should default to true for agents"
);
}
#[test]
fn corporate_identity_requires_complete_verifier_config() {
let _guard = ENV_MUTEX.lock().unwrap();
clear_corporate_identity_env();
std::env::set_var("BUZZ_REQUIRE_CORPORATE_IDENTITY", "true");
let err = Config::from_env().expect_err("incomplete corporate identity config");
let msg = err.to_string();
clear_corporate_identity_env();
assert!(msg.contains("BUZZ_CORPORATE_IDENTITY_JWKS_URI"));
assert!(msg.contains("BUZZ_CORPORATE_IDENTITY_ISSUER"));
assert!(msg.contains("BUZZ_CORPORATE_IDENTITY_AUDIENCE"));
}
#[test]
fn corporate_identity_config_can_be_enabled() {
let _guard = ENV_MUTEX.lock().unwrap();
clear_corporate_identity_env();
std::env::set_var("BUZZ_REQUIRE_CORPORATE_IDENTITY", "true");
std::env::set_var(
"BUZZ_CORPORATE_IDENTITY_JWKS_URI",
"https://idp.example/.well-known/jwks.json",
);
std::env::set_var("BUZZ_CORPORATE_IDENTITY_ISSUER", "https://idp.example");
std::env::set_var("BUZZ_CORPORATE_IDENTITY_AUDIENCE", "buzz-relay");
std::env::set_var("BUZZ_CORPORATE_IDENTITY_UID_CLAIM", "employee_id");
std::env::set_var("BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM", "email");
std::env::set_var("BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM", "buzz_npub");
let config = Config::from_env().expect("corporate identity config");
clear_corporate_identity_env();
assert!(config.corporate_identity.require);
assert_eq!(config.corporate_identity.uid_claim, "employee_id");
assert_eq!(
config.corporate_identity.npub_claim.as_deref(),
Some("buzz_npub")
);
}
#[test]

17
crates/buzz-relay/src/connection.rs
View File

@@ -59,6 +59,8 @@ pub struct ConnectionState {
pub tenant: TenantContext,
/// Remote socket address of the client.
pub remote_addr: SocketAddr,
/// Optional corporate identity JWT captured from the WebSocket upgrade request.
pub corporate_identity_jwt: Option<String>,
/// Current NIP-42 authentication state.
pub auth_state: RwLock<AuthState>,
/// Active subscriptions keyed by subscription ID.
@@ -120,6 +122,7 @@ pub async fn handle_connection(
state: Arc<AppState>,
addr: SocketAddr,
tenant: TenantContext,
corporate_identity_jwt: Option<String>,
) {
let conn_id = Uuid::new_v4();
let cancel = CancellationToken::new();
@@ -133,7 +136,17 @@ pub async fn handle_connection(
community_id,
cancel.clone(),
move || async move { check_state.db.is_community_active(community_id).await },
move || handle_active_connection(socket, run_state, addr, tenant, conn_id, cancel),
move || {
handle_active_connection(
socket,
run_state,
addr,
tenant,
conn_id,
cancel,
corporate_identity_jwt,
)
},
)
.await;
}
@@ -145,6 +158,7 @@ async fn handle_active_connection(
tenant: TenantContext,
conn_id: Uuid,
cancel: CancellationToken,
corporate_identity_jwt: Option<String>,
) {
let permit = match state.conn_semaphore.clone().try_acquire_owned() {
Ok(p) => p,
@@ -168,6 +182,7 @@ async fn handle_active_connection(
conn_id,
tenant,
remote_addr: addr,
corporate_identity_jwt,
auth_state: RwLock::new(AuthState::Pending {
challenge: challenge.clone(),
}),

801
crates/buzz-relay/src/corporate_identity.rs Normal file
View File

@@ -0,0 +1,801 @@
//! Corporate identity verification and uid/pubkey binding.
//!
//! This module is intentionally relay-local. `buzz-auth` remains the generic
//! Nostr proof layer; corporate identity is deployment policy layered after a
//! request proves control of a Nostr key.
use std::sync::Arc;
use std::time::{Duration, Instant};
use axum::{
http::{HeaderMap, StatusCode},
response::Json,
};
use jsonwebtoken::{
decode, decode_header,
jwk::{Jwk, JwkSet},
Algorithm, DecodingKey, Validation,
};
use nostr::{FromBech32, PublicKey};
use serde::Deserialize;
use serde_json::{Map, Value};
use thiserror::Error;
use tokio::sync::RwLock;
use tracing::{debug, warn};
use buzz_core::CommunityId;
use buzz_db::identity_binding::{BindIdentityResult, SOURCE_DB_BINDING, SOURCE_JWT_NPUB};
use crate::config::CorporateIdentityConfig;
use crate::state::AppState;
const JWKS_CACHE_TTL: Duration = Duration::from_secs(300);
#[derive(Debug, Clone)]
struct CachedJwks {
set: JwkSet,
expires_at: Instant,
}
/// Validated corporate identity claims used by Buzz.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CorporateJwtClaims {
/// Stable corporate uid claim.
pub uid: String,
/// Human-readable verified identity claim.
pub display_name: String,
/// Optional pubkey carried by the IdP.
pub pubkey: Option<PublicKey>,
}
#[derive(Debug, Deserialize)]
struct RawJwtClaims {
#[serde(flatten)]
claims: Map<String, Value>,
}
/// Service that verifies corporate identity JWTs against configured JWKS.
#[derive(Debug)]
pub struct CorporateIdentityService {
config: CorporateIdentityConfig,
http: reqwest::Client,
jwks: RwLock<Option<CachedJwks>>,
}
impl CorporateIdentityService {
/// Build a corporate identity verifier from relay config.
pub fn new(config: CorporateIdentityConfig) -> Self {
Self {
config,
http: reqwest::Client::new(),
jwks: RwLock::new(None),
}
}
/// Validate a JWT and extract the configured corporate identity claims.
pub async fn validate_jwt(
&self,
token: &str,
) -> Result<CorporateJwtClaims, CorporateIdentityError> {
let header = decode_header(token)
.map_err(|e| CorporateIdentityError::InvalidJwt(format!("invalid JWT header: {e}")))?;
if !is_allowed_jwt_algorithm(header.alg) {
return Err(CorporateIdentityError::InvalidJwt(format!(
"unsupported JWT algorithm: {:?}",
header.alg
)));
}
let kid = header
.kid
.as_deref()
.ok_or(CorporateIdentityError::MissingKid)?;
let jwk = self.jwk_for_kid(kid).await?;
let decoding_key = DecodingKey::from_jwk(&jwk).map_err(|e| {
CorporateIdentityError::InvalidJwt(format!("invalid JWK for kid {kid}: {e}"))
})?;
let mut validation = Validation::new(header.alg);
validation.set_issuer(&[self.config.issuer.as_str()]);
validation.set_audience(&[self.config.audience.as_str()]);
let decoded = decode::<RawJwtClaims>(token, &decoding_key, &validation)
.map_err(|e| CorporateIdentityError::InvalidJwt(e.to_string()))?;
let uid = claim_string(&decoded.claims.claims, &self.config.uid_claim)?;
let display_name = claim_string(&decoded.claims.claims, &self.config.display_claim)?;
let pubkey =
optional_pubkey_claim(&decoded.claims.claims, self.config.npub_claim.as_deref())?;
Ok(CorporateJwtClaims {
uid,
display_name,
pubkey,
})
}
async fn jwk_for_kid(&self, kid: &str) -> Result<Jwk, CorporateIdentityError> {
let now = Instant::now();
{
let cache = self.jwks.read().await;
if let Some(cached) = cache.as_ref() {
if cached.expires_at > now {
if let Some(jwk) = cached.set.find(kid) {
return Ok(jwk.clone());
}
return Err(CorporateIdentityError::Jwks(format!(
"kid not found in fresh JWKS cache: {kid}"
)));
}
}
}
let set = self.fetch_jwks().await?;
let jwk = set.find(kid).cloned();
*self.jwks.write().await = Some(CachedJwks {
set,
expires_at: Instant::now() + JWKS_CACHE_TTL,
});
jwk.ok_or_else(|| CorporateIdentityError::Jwks(format!("kid not found: {kid}")))
}
async fn fetch_jwks(&self) -> Result<JwkSet, CorporateIdentityError> {
let response = self
.http
.get(&self.config.jwks_uri)
.send()
.await
.map_err(|e| CorporateIdentityError::Jwks(e.to_string()))?
.error_for_status()
.map_err(|e| CorporateIdentityError::Jwks(e.to_string()))?;
response
.json::<JwkSet>()
.await
.map_err(|e| CorporateIdentityError::Jwks(e.to_string()))
}
}
/// Outcome of corporate identity enforcement.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CorporateIdentityDecision {
/// Corporate identity is disabled for this relay.
NotRequired,
/// The signer authenticated directly with a corporate identity JWT.
Direct {
/// Stable corporate uid claim.
uid: String,
/// Verified display claim.
display_name: String,
/// Binding operation outcome.
binding: BindIdentityResult,
},
/// The signer is an agent admitted through a bound owner pubkey.
Delegated {
/// NIP-OA owner pubkey that already has an active corporate binding.
owner_pubkey: PublicKey,
},
}
/// Errors produced by corporate identity verification.
#[derive(Debug, Error)]
pub enum CorporateIdentityError {
/// No JWT was available and delegation did not apply.
#[error("corporate identity JWT missing")]
MissingJwt,
/// JWT header did not include a `kid`.
#[error("corporate identity JWT missing kid")]
MissingKid,
/// JWT signature or claims failed validation.
#[error("invalid corporate identity JWT: {0}")]
InvalidJwt(String),
/// JWKS fetch or lookup failed.
#[error("corporate identity JWKS unavailable: {0}")]
Jwks(String),
/// A configured claim is missing or not a string.
#[error("invalid corporate identity claim {claim}: {reason}")]
InvalidClaim {
/// Claim name.
claim: String,
/// Validation reason.
reason: String,
},
/// The IdP-provided pubkey does not match the authenticated signer.
#[error("corporate identity npub claim does not match authenticated signer")]
NpubMismatch,
/// The requested uid/pubkey binding conflicts with an active binding.
#[error("corporate identity binding conflict")]
BindingConflict,
/// NIP-OA delegation was present but did not satisfy corporate identity.
#[error("corporate identity delegation denied")]
DelegationDenied,
/// Database operation failed.
#[error("corporate identity database error: {0}")]
Db(#[from] buzz_db::DbError),
}
impl CorporateIdentityError {
/// HTTP status appropriate for this error.
pub fn status_code(&self) -> StatusCode {
match self {
Self::MissingJwt | Self::MissingKid | Self::InvalidJwt(_) | Self::Jwks(_) => {
StatusCode::UNAUTHORIZED
}
Self::InvalidClaim { .. }
| Self::NpubMismatch
| Self::BindingConflict
| Self::DelegationDenied => StatusCode::FORBIDDEN,
Self::Db(_) => StatusCode::INTERNAL_SERVER_ERROR,
}
}
/// Sanitized message safe to return to clients.
pub fn public_message(&self) -> &'static str {
match self {
Self::MissingJwt => "corporate identity required",
Self::MissingKid | Self::InvalidJwt(_) | Self::Jwks(_) => {
"corporate identity verification failed"
}
Self::InvalidClaim { .. } => "corporate identity claim invalid",
Self::NpubMismatch => "corporate identity pubkey mismatch",
Self::BindingConflict => "corporate identity binding conflict",
Self::DelegationDenied => "corporate identity delegation denied",
Self::Db(_) => "corporate identity unavailable",
}
}
/// Convert to the standard API error shape.
pub fn into_api_error(self) -> (StatusCode, Json<Value>) {
let status = self.status_code();
let message = self.public_message();
if status.is_server_error() {
warn!(error = %self, "corporate identity enforcement failed");
}
(status, Json(serde_json::json!({ "error": message })))
}
}
/// Extract a corporate identity JWT from the configured request header.
pub fn identity_jwt_from_headers(
headers: &HeaderMap,
config: &CorporateIdentityConfig,
) -> Option<String> {
headers
.get(config.jwt_header.as_str())
.and_then(|v| v.to_str().ok())
.map(str::trim)
.and_then(|raw| {
raw.strip_prefix("Bearer ")
.unwrap_or(raw)
.trim()
.split(',')
.next()
})
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
}
/// Enforce corporate identity for an already NIP-authenticated signer.
pub async fn enforce_corporate_identity(
state: &AppState,
community_id: CommunityId,
signer: PublicKey,
identity_jwt: Option<&str>,
auth_tag_json: Option<&str>,
) -> Result<CorporateIdentityDecision, CorporateIdentityError> {
let result =
enforce_corporate_identity_inner(state, community_id, signer, identity_jwt, auth_tag_json)
.await;
if let Err(error) = &result {
record_corporate_identity_denial(error);
}
result
}
async fn enforce_corporate_identity_inner(
state: &AppState,
community_id: CommunityId,
signer: PublicKey,
identity_jwt: Option<&str>,
auth_tag_json: Option<&str>,
) -> Result<CorporateIdentityDecision, CorporateIdentityError> {
let Some(service) = state.corporate_identity.as_ref() else {
return Ok(CorporateIdentityDecision::NotRequired);
};
if let Some(token) = identity_jwt {
let claims = service.validate_jwt(token).await?;
let source = binding_source_for_signer(claims.pubkey, signer)?;
let binding = state
.db
.bind_or_validate_identity(
community_id,
&claims.uid,
signer.as_bytes(),
Some(&claims.display_name),
source,
)
.await?;
let binding = match binding {
BindIdentityResult::Conflict(conflict) => {
metrics::counter!("buzz_corporate_identity_bindings_total", "result" => "conflict")
.increment(1);
record_identity_binding_audit(
state,
community_id,
buzz_audit::AuditAction::CorporateIdentityBindingConflict,
signer,
&claims.uid,
serde_json::json!({
"source": source,
"existing_uid": conflict.uid,
"existing_pubkey": hex::encode(conflict.pubkey),
"existing_source": conflict.source,
}),
)
.await;
warn!(
uid = %claims.uid,
signer = %signer.to_hex(),
"corporate identity binding conflict"
);
return Err(CorporateIdentityError::BindingConflict);
}
binding => binding,
};
record_identity_binding_metric(&binding);
if matches!(binding, BindIdentityResult::Created) {
record_identity_binding_audit(
state,
community_id,
buzz_audit::AuditAction::CorporateIdentityBindingCreated,
signer,
&claims.uid,
serde_json::json!({ "source": source }),
)
.await;
}
debug!(
uid = %claims.uid,
signer = %signer.to_hex(),
source,
"corporate identity verified"
);
return Ok(CorporateIdentityDecision::Direct {
uid: claims.uid,
display_name: claims.display_name,
binding,
});
}
enforce_delegated_corporate_identity(
&state.db,
&service.config,
community_id,
signer,
auth_tag_json,
)
.await
}
async fn enforce_delegated_corporate_identity(
db: &buzz_db::Db,
config: &CorporateIdentityConfig,
community_id: CommunityId,
signer: PublicKey,
auth_tag_json: Option<&str>,
) -> Result<CorporateIdentityDecision, CorporateIdentityError> {
if config.allow_delegation {
if let Some(owner_pubkey) =
crate::api::relay_members::extract_nip_oa_owner(signer.as_bytes(), auth_tag_json)
{
let owner_binding = db
.get_active_identity_binding_by_pubkey(community_id, owner_pubkey.as_bytes())
.await?;
if owner_binding.is_some() {
debug!(
agent = %signer.to_hex(),
owner = %owner_pubkey.to_hex(),
"corporate identity granted via NIP-OA owner binding"
);
return Ok(CorporateIdentityDecision::Delegated { owner_pubkey });
}
}
}
if auth_tag_json.is_some() {
Err(CorporateIdentityError::DelegationDenied)
} else {
Err(CorporateIdentityError::MissingJwt)
}
}
fn is_allowed_jwt_algorithm(algorithm: Algorithm) -> bool {
matches!(
algorithm,
Algorithm::RS256
| Algorithm::RS384
| Algorithm::RS512
| Algorithm::PS256
| Algorithm::PS384
| Algorithm::PS512
| Algorithm::ES256
| Algorithm::ES384
| Algorithm::EdDSA
)
}
fn binding_source_for_signer(
claim_pubkey: Option<PublicKey>,
signer: PublicKey,
) -> Result<&'static str, CorporateIdentityError> {
match claim_pubkey {
Some(claim_pubkey) => {
if claim_pubkey != signer {
warn!(
signer = %signer.to_hex(),
claim_pubkey = %claim_pubkey.to_hex(),
"corporate identity JWT npub claim does not match signer"
);
return Err(CorporateIdentityError::NpubMismatch);
}
Ok(SOURCE_JWT_NPUB)
}
None => Ok(SOURCE_DB_BINDING),
}
}
fn claim_string(
claims: &Map<String, Value>,
claim: &str,
) -> Result<String, CorporateIdentityError> {
let value = claims
.get(claim)
.ok_or_else(|| CorporateIdentityError::InvalidClaim {
claim: claim.to_string(),
reason: "missing".to_string(),
})?;
let value = value
.as_str()
.map(str::trim)
.filter(|s| !s.is_empty())
.ok_or_else(|| CorporateIdentityError::InvalidClaim {
claim: claim.to_string(),
reason: "must be a non-empty string".to_string(),
})?;
Ok(value.to_string())
}
fn optional_claim_string(
claims: &Map<String, Value>,
claim: &str,
) -> Result<Option<String>, CorporateIdentityError> {
let Some(value) = claims.get(claim) else {
return Ok(None);
};
let value = value
.as_str()
.map(str::trim)
.filter(|s| !s.is_empty())
.ok_or_else(|| CorporateIdentityError::InvalidClaim {
claim: claim.to_string(),
reason: "must be a non-empty string".to_string(),
})?;
Ok(Some(value.to_string()))
}
fn optional_pubkey_claim(
claims: &Map<String, Value>,
claim: Option<&str>,
) -> Result<Option<PublicKey>, CorporateIdentityError> {
match claim {
Some(claim) => optional_claim_string(claims, claim)?
.as_deref()
.map(|raw| parse_pubkey_claim(claim, raw))
.transpose(),
None => Ok(None),
}
}
fn parse_pubkey_claim(claim: &str, value: &str) -> Result<PublicKey, CorporateIdentityError> {
if value.starts_with("npub1") {
PublicKey::from_bech32(value).map_err(|e| CorporateIdentityError::InvalidClaim {
claim: claim.to_string(),
reason: format!("invalid npub: {e}"),
})
} else {
PublicKey::from_hex(value).map_err(|e| CorporateIdentityError::InvalidClaim {
claim: claim.to_string(),
reason: format!("invalid pubkey hex: {e}"),
})
}
}
/// Create an optional service from config.
pub fn service_from_config(
config: &CorporateIdentityConfig,
) -> Option<Arc<CorporateIdentityService>> {
config
.require
.then(|| Arc::new(CorporateIdentityService::new(config.clone())))
}
fn record_identity_binding_metric(binding: &BindIdentityResult) {
let result = match binding {
BindIdentityResult::Created => "created",
BindIdentityResult::Matched => "matched",
BindIdentityResult::Conflict(_) => "conflict",
};
metrics::counter!("buzz_corporate_identity_bindings_total", "result" => result).increment(1);
}
fn record_corporate_identity_denial(error: &CorporateIdentityError) {
let reason = match error {
CorporateIdentityError::MissingJwt => "missing_jwt",
CorporateIdentityError::MissingKid => "missing_kid",
CorporateIdentityError::InvalidJwt(_) => "invalid_jwt",
CorporateIdentityError::Jwks(_) => "jwks",
CorporateIdentityError::InvalidClaim { .. } => "invalid_claim",
CorporateIdentityError::NpubMismatch => "npub_mismatch",
CorporateIdentityError::BindingConflict => "binding_conflict",
CorporateIdentityError::DelegationDenied => "delegation_denied",
CorporateIdentityError::Db(_) => "db",
};
metrics::counter!("buzz_auth_failures_total", "reason" => "corporate_identity_denied")
.increment(1);
metrics::counter!("buzz_corporate_identity_denials_total", "reason" => reason).increment(1);
}
async fn record_identity_binding_audit(
state: &AppState,
community_id: CommunityId,
action: buzz_audit::AuditAction,
actor: PublicKey,
uid: &str,
detail: serde_json::Value,
) {
let Some(audit_tx) = &state.audit_tx else {
return;
};
if let Err(e) = audit_tx
.send(buzz_audit::NewAuditEntry {
community_id,
action,
actor_pubkey: Some(actor.to_bytes().to_vec()),
object_id: Some(uid.to_string()),
detail,
})
.await
{
warn!("Corporate identity audit channel closed — entry lost: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::{HeaderMap, HeaderName, HeaderValue};
use jsonwebtoken::jwk::JwkSet;
use jsonwebtoken::{encode, EncodingKey, Header};
use nostr::Keys;
use sqlx::PgPool;
use uuid::Uuid;
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
fn test_config() -> CorporateIdentityConfig {
CorporateIdentityConfig {
require: true,
jwt_header: "x-buzz-identity-token".to_string(),
allow_delegation: true,
jwks_uri: "http://127.0.0.1:9/jwks".to_string(),
issuer: "https://idp.example".to_string(),
audience: "buzz-relay".to_string(),
uid_claim: "sub".to_string(),
display_claim: "email".to_string(),
npub_claim: Some("buzz_npub".to_string()),
}
}
#[test]
fn rejects_hmac_jwt_algorithms_in_allowlist() {
assert!(!is_allowed_jwt_algorithm(Algorithm::HS256));
assert!(!is_allowed_jwt_algorithm(Algorithm::HS384));
assert!(!is_allowed_jwt_algorithm(Algorithm::HS512));
assert!(is_allowed_jwt_algorithm(Algorithm::RS256));
}
#[tokio::test]
async fn validate_jwt_rejects_hs256_before_jwks_lookup() {
let service = CorporateIdentityService::new(test_config());
let mut header = Header::new(Algorithm::HS256);
header.kid = Some("hs256-kid".to_string());
let token = encode(
&header,
&serde_json::json!({
"iss": "https://idp.example",
"aud": "buzz-relay",
"sub": "user-1",
"email": "user@example.com",
}),
&EncodingKey::from_secret(b"test-secret"),
)
.expect("encode test jwt");
let err = service
.validate_jwt(&token)
.await
.expect_err("HS256 must be rejected");
assert!(matches!(err, CorporateIdentityError::InvalidJwt(_)));
}
#[test]
fn extracts_bearer_token_from_comma_list_header() {
let config = test_config();
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("x-buzz-identity-token"),
HeaderValue::from_static("Bearer token-a, Bearer token-b"),
);
assert_eq!(
identity_jwt_from_headers(&headers, &config).as_deref(),
Some("token-a")
);
}
#[test]
fn missing_required_claim_is_invalid() {
let claims = Map::new();
let err = claim_string(&claims, "sub").expect_err("missing claim");
assert!(matches!(
err,
CorporateIdentityError::InvalidClaim { ref claim, .. } if claim == "sub"
));
}
#[test]
fn configured_npub_claim_is_optional_but_malformed_value_is_invalid() {
let mut claims = Map::new();
assert_eq!(
optional_pubkey_claim(&claims, Some("buzz_npub")).expect("missing is optional"),
None
);
claims.insert(
"buzz_npub".to_string(),
Value::String("not-an-npub".to_string()),
);
let err = optional_pubkey_claim(&claims, Some("buzz_npub"))
.expect_err("present malformed claim must fail");
assert!(matches!(
err,
CorporateIdentityError::InvalidClaim { ref claim, .. } if claim == "buzz_npub"
));
}
#[test]
fn npub_claim_must_match_authenticated_signer() {
let signer = Keys::generate().public_key();
let other = Keys::generate().public_key();
assert!(matches!(
binding_source_for_signer(Some(other), signer),
Err(CorporateIdentityError::NpubMismatch)
));
assert_eq!(
binding_source_for_signer(Some(signer), signer).expect("match"),
SOURCE_JWT_NPUB
);
assert_eq!(
binding_source_for_signer(None, signer).expect("db fallback"),
SOURCE_DB_BINDING
);
}
#[tokio::test]
async fn fresh_jwks_cache_miss_does_not_refetch() {
let service = CorporateIdentityService::new(test_config());
*service.jwks.write().await = Some(CachedJwks {
set: JwkSet { keys: Vec::new() },
expires_at: Instant::now() + Duration::from_secs(60),
});
let err = service
.jwk_for_kid("attacker-controlled-kid")
.await
.expect_err("fresh cache miss should fail without network fetch");
assert!(matches!(
err,
CorporateIdentityError::Jwks(ref msg) if msg.contains("fresh JWKS cache")
));
}
async fn setup_db() -> (buzz_db::Db, PgPool) {
let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
.unwrap_or_else(|_| TEST_DB_URL.to_owned());
let pool = PgPool::connect(&database_url)
.await
.expect("connect to test DB");
let db = buzz_db::Db::from_pool(pool.clone());
db.migrate().await.expect("run migrations");
(db, pool)
}
async fn make_community(pool: &PgPool) -> CommunityId {
let id = Uuid::new_v4();
let host = format!("corporate-identity-test-{}.example", id.simple());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(host)
.execute(pool)
.await
.expect("insert test community");
CommunityId::from_uuid(id)
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn delegation_requires_owner_identity_binding() {
let (db, pool) = setup_db().await;
let community = make_community(&pool).await;
let owner_keys = Keys::generate();
let agent_keys = Keys::generate();
let agent_pubkey = agent_keys.public_key();
let auth_tag = buzz_sdk::nip_oa::compute_auth_tag(&owner_keys, &agent_pubkey, "").unwrap();
let config = test_config();
let err = enforce_delegated_corporate_identity(
&db,
&config,
community,
agent_pubkey,
Some(&auth_tag),
)
.await
.expect_err("owner without binding should be denied");
assert!(matches!(err, CorporateIdentityError::DelegationDenied));
db.bind_or_validate_identity(
community,
"owner-uid",
owner_keys.public_key().as_bytes(),
Some("owner@example.com"),
SOURCE_DB_BINDING,
)
.await
.expect("create owner binding");
let decision = enforce_delegated_corporate_identity(
&db,
&config,
community,
agent_pubkey,
Some(&auth_tag),
)
.await
.expect("owner binding admits agent");
assert_eq!(
decision,
CorporateIdentityDecision::Delegated {
owner_pubkey: owner_keys.public_key()
}
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn missing_jwt_without_auth_tag_is_missing_jwt() {
let (db, pool) = setup_db().await;
let community = make_community(&pool).await;
let signer = Keys::generate().public_key();
let config = test_config();
let err = enforce_delegated_corporate_identity(&db, &config, community, signer, None)
.await
.expect_err("no JWT and no delegation tag");
assert!(matches!(err, CorporateIdentityError::MissingJwt));
}
}

27
crates/buzz-relay/src/handlers/auth.rs
View File

@@ -183,6 +183,33 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
}
}
match crate::corporate_identity::enforce_corporate_identity(
&state,
conn.tenant.community(),
pubkey,
conn.corporate_identity_jwt.as_deref(),
auth_tag_json.as_deref(),
)
.await
{
Ok(crate::corporate_identity::CorporateIdentityDecision::Delegated {
owner_pubkey,
}) => {
auth_ctx.agent_owner_pubkey = Some(owner_pubkey);
}
Ok(_) => {}
Err(e) => {
warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), error = %e, "corporate identity denied");
*conn.auth_state.write().await = AuthState::Failed;
conn.send(RelayMessage::ok(
&event_id_hex,
false,
&format!("restricted: {}", e.public_message()),
));
return;
}
}
// Pubkey allowlist gate — only for pubkey-only auth.
if state.config.pubkey_allowlist_enabled
&& auth_ctx.auth_method == buzz_auth::AuthMethod::Nip42

1
crates/buzz-relay/src/handlers/event.rs
View File

@@ -1386,6 +1386,7 @@ mod tests {
conn_id: Uuid::new_v4(),
tenant: buzz_core::TenantContext::resolved(community_b, "b.example"),
remote_addr: "127.0.0.1:1234".parse().expect("socket addr"),
corporate_identity_jwt: None,
auth_state: RwLock::new(crate::connection::AuthState::Authenticated(
buzz_auth::AuthContext {
pubkey: agent.public_key(),

2
crates/buzz-relay/src/lib.rs
View File

@@ -17,6 +17,8 @@ pub mod config;
pub mod conformance;
/// WebSocket connection lifecycle and state.
pub mod connection;
/// Corporate identity verification and uid/pubkey binding.
pub mod corporate_identity;
/// Relay error types.
pub mod error;
/// WebSocket message handlers for NIP-01 client commands.

8
crates/buzz-relay/src/router.rs
View File

@@ -310,6 +310,10 @@ async fn nip11_or_ws_handler(
.into_response();
}
};
let corporate_identity_jwt = crate::corporate_identity::identity_jwt_from_headers(
&headers,
&state.config.corporate_identity,
);
let max_frame_bytes = state.config.max_frame_bytes;
match WebSocketUpgrade::from_request(req, &state).await {
@@ -324,7 +328,9 @@ async fn nip11_or_ws_handler(
return (StatusCode::SERVICE_UNAVAILABLE, "relay restarting").into_response();
}
limit_relay_websocket(ws, max_frame_bytes)
.on_upgrade(move |socket| handle_connection(socket, state, addr, tenant))
.on_upgrade(move |socket| {
handle_connection(socket, state, addr, tenant, corporate_identity_jwt)
})
.into_response()
}
Err(_) => {

7
crates/buzz-relay/src/state.rs
View File

@@ -32,6 +32,7 @@ use deadpool_redis;
use crate::audio::AudioRoomManager;
use crate::config::Config;
use crate::connection::ConnectionSubscriptions;
use crate::corporate_identity::CorporateIdentityService;
use crate::subscription::SubscriptionRegistry;
pub(crate) type ScopedPubkeyKey = (CommunityId, [u8; 32]);
@@ -498,6 +499,8 @@ pub struct AppState {
pub pubsub: Arc<PubSubManager>,
/// Authentication service.
pub auth: Arc<AuthService>,
/// Optional corporate identity verifier.
pub corporate_identity: Option<Arc<CorporateIdentityService>>,
/// Full-text search service.
pub search: Arc<SearchService>,
/// Registry of active client subscriptions.
@@ -649,6 +652,8 @@ impl AppState {
let max_connections = config.max_connections;
let max_concurrent_handlers = config.max_concurrent_handlers;
let search_arc = Arc::new(search);
let corporate_identity =
crate::corporate_identity::service_from_config(&config.corporate_identity);
let audit_arc = audit.into().map(Arc::new);
let (audit_tx, mut audit_rx) = mpsc::channel::<buzz_audit::NewAuditEntry>(1000);
@@ -719,6 +724,7 @@ impl AppState {
audit: audit_arc,
pubsub,
auth: Arc::new(auth),
corporate_identity,
search: search_arc,
sub_registry: Arc::new(SubscriptionRegistry::new()),
conn_manager: Arc::new(ConnectionManager::new()),
@@ -1357,6 +1363,7 @@ mod tests {
"test.local".to_string(),
),
remote_addr: "127.0.0.1:1234".parse().unwrap(),
corporate_identity_jwt: None,
auth_state: RwLock::new(AuthState::Failed),
subscriptions: Arc::new(Mutex::new(HashMap::new())),
send_tx: tx.clone(),

35
migrations/0027_identity_bindings.sql Normal file
View File

@@ -0,0 +1,35 @@
-- Corporate identity bindings.
--
-- This is the relay-side foundation for mapping a corporate IdP subject to a
-- Nostr pubkey. It is intentionally not a full grant/session model: lifecycle
-- operations such as admin revocation, rotation workflows, and live connection
-- eviction are follow-up work, but the columns/indexes below preserve those
-- states without requiring a later destructive schema rewrite.
CREATE TABLE identity_bindings (
community_id UUID NOT NULL REFERENCES communities(id),
uid TEXT NOT NULL,
pubkey BYTEA NOT NULL,
display_name TEXT,
source TEXT NOT NULL CHECK (source IN ('jwt_npub', 'db_binding')),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
revoked_at TIMESTAMPTZ,
revoked_by BYTEA,
revoked_reason TEXT,
CONSTRAINT chk_identity_bindings_uid_not_empty CHECK (length(uid) > 0),
CONSTRAINT chk_identity_bindings_pubkey_len CHECK (length(pubkey) = 32),
CONSTRAINT chk_identity_bindings_revoked_by_len CHECK (revoked_by IS NULL OR length(revoked_by) = 32)
);
CREATE UNIQUE INDEX idx_identity_bindings_active_uid
ON identity_bindings (community_id, uid)
WHERE revoked_at IS NULL;
CREATE UNIQUE INDEX idx_identity_bindings_active_pubkey
ON identity_bindings (community_id, pubkey)
WHERE revoked_at IS NULL;
CREATE INDEX idx_identity_bindings_pubkey
ON identity_bindings (community_id, pubkey);

34
schema/schema.sql
View File

@@ -184,6 +184,40 @@ CREATE UNIQUE INDEX idx_users_nip05 ON users (community_id, lower(nip05_handle))
CREATE UNIQUE INDEX idx_users_okta ON users (community_id, okta_user_id)
WHERE okta_user_id IS NOT NULL;
-- ── Corporate identity bindings ──────────────────────────────────────────────
-- Conformance: corporate identity is community-scoped. A corporate uid is the
-- stable product/user-management identity; a Nostr pubkey is the protocol
-- credential currently bound to it. This table is intentionally a binding
-- foundation, not a full grant/session lifecycle model. Revocation columns are
-- reserved for follow-up admin/rotation flows, while PR1 only creates or
-- validates active bindings during existing auth paths.
CREATE TABLE identity_bindings (
community_id UUID NOT NULL REFERENCES communities(id),
uid TEXT NOT NULL,
pubkey BYTEA NOT NULL,
display_name TEXT,
source TEXT NOT NULL CHECK (source IN ('jwt_npub', 'db_binding')),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
revoked_at TIMESTAMPTZ,
revoked_by BYTEA,
revoked_reason TEXT,
CONSTRAINT chk_identity_bindings_uid_not_empty CHECK (length(uid) > 0),
CONSTRAINT chk_identity_bindings_pubkey_len CHECK (length(pubkey) = 32),
CONSTRAINT chk_identity_bindings_revoked_by_len CHECK (revoked_by IS NULL OR length(revoked_by) = 32)
);
CREATE UNIQUE INDEX idx_identity_bindings_active_uid
ON identity_bindings (community_id, uid)
WHERE revoked_at IS NULL;
CREATE UNIQUE INDEX idx_identity_bindings_active_pubkey
ON identity_bindings (community_id, pubkey)
WHERE revoked_at IS NULL;
CREATE INDEX idx_identity_bindings_pubkey
ON identity_bindings (community_id, pubkey);
-- ── Events (partitioned by month on created_at) ──────────────────────────────
-- Conformance: "Channel-less global events and DMs". `community_id` leads the
-- PK and every hot-path index. Partition stays BY RANGE (created_at) — the