Kanidm group-gated rooms and minimal mesh calling

Rooms can now require a Kanidm group (via the `groups` OIDC claim,
mapped by `oauth2 update-claim-map` server-side) - dev/ops require
`developers`, enforced at every message path (send, history, SSE).

Adds a minimal WebRTC mesh call feature scoped to the lobby room,
signaled over a separate `call.room.*` NATS subject kept out of the
chat archive: public STUN only, no TURN, no SFU - small groups on
friendly networks, by design.

src/server/oidc.rs

@@ -185,10 +185,20 @@ pub async fn callback(
         .map(|n| n.as_str().to_string())
         .unwrap_or_else(|| username.clone());
 
+    // `groups` is a custom claim (Kanidm `oauth2 update-claim-map`), not
+    // something the Core* typed claims struct above knows about. The
+    // signature is already verified by `id_token.claims(...)` above, so
+    // re-reading the same payload's raw JSON for one more field is safe -
+    // just a plain field extraction, not a second verification step.
+    // IdToken's Serialize impl (not Display - it has none) produces the
+    // raw compact JWT string "header.payload.signature".
+    let groups = extract_groups_claim(&id_token);
+
     let user = User {
         sub: claims.subject().as_str().to_string(),
         username,
         display_name,
+        groups,
     };
 
     // Rotate the session id on privilege change, then store the user.
@@ -207,3 +217,34 @@ pub async fn logout(session: Session) -> Result<Redirect, HandlerError> {
     session.flush().await.map_err(internal)?;
     Ok(Redirect::to("/"))
 }
+
+/// Pulls the `groups` custom claim (Kanidm `oauth2 update-claim-map`) out
+/// of an ID token's raw JWT payload. `IdToken`'s `Serialize` impl (it has
+/// no `Display`) produces the compact "header.payload.signature" string,
+/// which is where this reads from - the signature itself is never
+/// re-checked here, that already happened via `id_token.claims(...)`
+/// before this is called. Defensive by design: any parse failure (no
+/// claim, wrong shape) just yields no groups rather than failing login.
+fn extract_groups_claim<T: serde::Serialize>(id_token: &T) -> Vec<String> {
+    use base64::Engine;
+
+    let Ok(serde_json::Value::String(compact)) = serde_json::to_value(id_token) else {
+        return Vec::new();
+    };
+    let Some(payload_b64) = compact.split('.').nth(1) else {
+        return Vec::new();
+    };
+    let Ok(payload_bytes) = base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(payload_b64)
+    else {
+        return Vec::new();
+    };
+    let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&payload_bytes) else {
+        return Vec::new();
+    };
+
+    payload
+        .get("groups")
+        .and_then(|g| g.as_array())
+        .map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+        .unwrap_or_default()
+}

src/server/sse.rs

@@ -12,28 +12,45 @@ use futures::{Stream, StreamExt};
 use tower_sessions::Session;
 
 use crate::auth::{User, SESSION_USER_KEY};
-use crate::chat::{is_valid_room, room_subject};
+use crate::call::call_subject;
+use crate::chat::{is_authorized_for_room, is_valid_room, room_subject};
 
 use super::AppState;
 
-/// GET /sse/{room} — stream the room's NATS subject to the browser.
+/// Shared by `room_events` and `call_events`: signed in, valid room, and
+/// authorized for it (Kanidm group gate, checked against the session's
+/// own `groups` - see `chat::is_authorized_for_room`). Note this is only
+/// checked once, at connect time - a long-lived SSE stream doesn't get
+/// re-checked if the user's groups change mid-connection (same kind of
+/// staleness the "still signed in at all" check already has).
+async fn authorize_room_stream(
+    room: &str,
+    session: &Session,
+) -> Result<(), (StatusCode, &'static str)> {
+    let user = session
+        .get::<User>(SESSION_USER_KEY)
+        .await
+        .ok()
+        .flatten();
+    let Some(user) = user else {
+        return Err((StatusCode::UNAUTHORIZED, "sign in first"));
+    };
+    if !is_valid_room(room) {
+        return Err((StatusCode::NOT_FOUND, "unknown room"));
+    }
+    if !is_authorized_for_room(&user, room) {
+        return Err((StatusCode::FORBIDDEN, "not authorized for this room"));
+    }
+    Ok(())
+}
+
+/// GET /sse/{room} — stream the room's chat NATS subject to the browser.
 pub async fn room_events(
     Path(room): Path<String>,
     State(state): State<AppState>,
     session: Session,
 ) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, &'static str)> {
-    let signed_in = session
-        .get::<User>(SESSION_USER_KEY)
-        .await
-        .ok()
-        .flatten()
-        .is_some();
-    if !signed_in {
-        return Err((StatusCode::UNAUTHORIZED, "sign in first"));
-    }
-    if !is_valid_room(&room) {
-        return Err((StatusCode::NOT_FOUND, "unknown room"));
-    }
+    authorize_room_stream(&room, &session).await?;
 
     let subscriber = state
         .nats
@@ -56,3 +73,37 @@ pub async fn room_events(
             .text("ping"),
     ))
 }
+
+/// GET /call-sse/{room} — stream the room's call-signaling NATS subject
+/// (SDP offers/answers, ICE candidates). Deliberately a separate subject
+/// namespace (`call.room.*`, not `chat.room.*`) so this never touches the
+/// JetStream/Postgres chat archive - ephemeral signaling has no business
+/// being durably stored.
+pub async fn call_events(
+    Path(room): Path<String>,
+    State(state): State<AppState>,
+    session: Session,
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, &'static str)> {
+    authorize_room_stream(&room, &session).await?;
+
+    let subscriber = state
+        .nats
+        .subscribe(call_subject(&room))
+        .await
+        .map_err(|e| {
+            tracing::error!("nats subscribe failed: {e}");
+            (StatusCode::BAD_GATEWAY, "message bus unavailable")
+        })?;
+
+    let stream = subscriber.map(|msg| {
+        Ok(Event::default()
+            .event("signal")
+            .data(String::from_utf8_lossy(&msg.payload).into_owned()))
+    });
+
+    Ok(Sse::new(stream).keep_alive(
+        KeepAlive::new()
+            .interval(Duration::from_secs(15))
+            .text("ping"),
+    ))
+}