fix(cluster): harden admin proof, dual-role drain, delete timeout

src/cluster/command.rs

@@ -74,22 +74,15 @@ pub enum ClusterResponse {
 }
 
 impl ClusterCommand {
-    /// Commands that may be proposed over the inter-node control plane.
-    pub fn allowed_on_control_plane(&self) -> bool {
-        let _ = self;
-        true
-    }
-
     /// Whether a `ClientWrite` over the control plane must carry an HTTP-API
-    /// `admin_proof`. Ownership acquire/release is excluded so nodes can
-    /// coordinate RTMP sessions without holding the admin bearer token.
+    /// `admin_proof`. All durable mutations require proof so a member that
+    /// only knows `CLUSTER_SECRET` cannot mutate replicated state (including
+    /// stream ownership). Local leaders still apply via `raft.client_write`
+    /// after their own HTTP/session path; followers mint proof from the
+    /// session-hook API token when forwarding.
     pub fn requires_admin_proof(&self) -> bool {
-        !matches!(
-            self,
-            ClusterCommand::AcquireStreamOwner { .. }
-                | ClusterCommand::ReleaseStreamOwner { .. }
-                | ClusterCommand::ReleaseOwnersForNode { .. }
-        )
+        let _ = self;
+        true
     }
 }
 
@@ -120,30 +113,49 @@ mod tests {
     use super::ClusterCommand;
 
     #[test]
-    fn control_plane_allows_member_forwarded_admin_writes() {
+    fn all_client_writes_require_admin_proof() {
         assert!(
             ClusterCommand::AcquireStreamOwner {
                 stream_id: "s".into(),
                 node_id: 1,
                 epoch: 1,
                 acquired_at: 0,
             }
-            .allowed_on_control_plane()
+            .requires_admin_proof()
         );
-        assert!(ClusterCommand::SetApiToken { token: "t".into() }.allowed_on_control_plane());
-    }
-
-    #[test]
-    fn ownership_commands_skip_admin_proof() {
         assert!(
-            !ClusterCommand::AcquireStreamOwner {
+            ClusterCommand::ReleaseStreamOwner {
                 stream_id: "s".into(),
-                node_id: 1,
                 epoch: 1,
-                acquired_at: 0,
             }
             .requires_admin_proof()
         );
+        assert!(
+            ClusterCommand::ReleaseOwnersForNode { node_id: 1 }.requires_admin_proof()
+        );
         assert!(ClusterCommand::SetApiToken { token: "t".into() }.requires_admin_proof());
+        assert!(
+            ClusterCommand::CreateStream {
+                stream: crate::db::Stream {
+                    id: "s".into(),
+                    name: "n".into(),
+                    app: "live".into(),
+                    publish_key: "p".into(),
+                    play_key: "k".into(),
+                    stats_key: "st".into(),
+                    enabled: true,
+                    created_at: 0,
+                },
+                default_viewer: crate::db::StreamViewer {
+                    id: "v".into(),
+                    stream_id: "s".into(),
+                    name: "default".into(),
+                    play_key: "k".into(),
+                    enabled: true,
+                    created_at: 0,
+                },
+            }
+            .requires_admin_proof()
+        );
     }
 }

src/cluster/manager.rs

@@ -696,11 +696,10 @@ impl ClusterManager {
     }
 
     fn block_on_write(&self, cmd: ClusterCommand) -> Result<ClusterResponse, CoordError> {
-        let proof = if cmd.requires_admin_proof() {
-            self.client_write_admin_proof(&cmd)?
-        } else {
-            String::new()
-        };
+        // Proof is required for every control-plane ClientWrite (including
+        // ownership). Local leaders still apply via raft.client_write below;
+        // the proof is only sent on ForwardToLeader.
+        let proof = self.client_write_admin_proof(&cmd)?;
         let raft = self.raft.clone();
         let secret = self.config.secret.clone();
         let local_id = self.config.node_id;
@@ -762,17 +761,25 @@ impl ClusterManager {
             self.admission.force_drain();
             return Ok(());
         }
-        self.forward_admin(node_id, network::ControlMessage::AdminDrain { node_id })
-            .await
+        let proof = self.admin_action_proof(&format!("AdminDrain:{node_id}"))?;
+        self.forward_admin(
+            node_id,
+            network::ControlMessage::AdminDrain { node_id, proof },
+        )
+        .await
     }
 
     pub async fn resume_node(&self, node_id: NodeId) -> Result<(), String> {
         if node_id == self.config.node_id {
             self.admission.force_resume();
             return Ok(());
         }
-        self.forward_admin(node_id, network::ControlMessage::AdminResume { node_id })
-            .await
+        let proof = self.admin_action_proof(&format!("AdminResume:{node_id}"))?;
+        self.forward_admin(
+            node_id,
+            network::ControlMessage::AdminResume { node_id, proof },
+        )
+        .await
     }
 
     pub async fn remove_peer(&self, node_id: NodeId) -> Result<(), String> {
@@ -851,14 +858,18 @@ impl ClusterManager {
         change: &ChangeMembers<NodeId, BasicNode>,
     ) -> Result<String, String> {
         let req = serde_json::to_string(change).map_err(|e| e.to_string())?;
+        self.admin_action_proof(&req)
+    }
+
+    fn admin_action_proof(&self, payload: &str) -> Result<String, String> {
         let token = self
             .session_hooks
             .lock()
             .as_ref()
             .map(|h| h.api_token.read().clone())
             .filter(|t| !t.is_empty())
-            .ok_or_else(|| "API token unavailable for membership change".to_string())?;
-        Ok(crate::cluster::security::admin_proof(&token, &req))
+            .ok_or_else(|| "API token unavailable for cluster admin action".to_string())?;
+        Ok(crate::cluster::security::admin_proof(&token, payload))
     }
 
     fn verify_admin_proof(&self, proof: &str, payload: &str) -> bool {
@@ -1008,11 +1019,15 @@ impl ClusterManager {
 
     fn handle_admin_control(&self, msg: network::ControlMessage) -> network::ControlMessage {
         match msg {
-            network::ControlMessage::AdminDrain { node_id } if node_id == self.config.node_id => {
+            network::ControlMessage::AdminDrain { node_id, .. }
+                if node_id == self.config.node_id =>
+            {
                 self.admission.force_drain();
                 network::ControlMessage::AdminOk
             }
-            network::ControlMessage::AdminResume { node_id } if node_id == self.config.node_id => {
+            network::ControlMessage::AdminResume { node_id, .. }
+                if node_id == self.config.node_id =>
+            {
                 self.admission.force_resume();
                 network::ControlMessage::AdminOk
             }
@@ -1022,13 +1037,29 @@ impl ClusterManager {
                     message: "use DELETE /api/v1/cluster/nodes/{id} on a voter".into(),
                 }
             }
+            // Network DrainStream/RevokeViewer are best-effort hints after Raft
+            // already applied the durable mutation. Ignore spoofed messages for
+            // streams/viewers that are still fully live in the local DB — Raft
+            // StateEffect still marks draining/revoked on apply.
             network::ControlMessage::DrainStream { stream_id } => {
-                self.mark_stream_draining(&stream_id);
-                network::ControlMessage::AdminOk
+                if self.stream_allows_network_drain(&stream_id) {
+                    self.mark_stream_draining(&stream_id);
+                    network::ControlMessage::AdminOk
+                } else {
+                    network::ControlMessage::AdminErr {
+                        message: "drain rejected: stream not pending delete".into(),
+                    }
+                }
             }
             network::ControlMessage::RevokeViewer { viewer_id } => {
-                self.mark_viewer_revoked(&viewer_id);
-                network::ControlMessage::AdminOk
+                if self.viewer_allows_network_revoke(&viewer_id) {
+                    self.mark_viewer_revoked(&viewer_id);
+                    network::ControlMessage::AdminOk
+                } else {
+                    network::ControlMessage::AdminErr {
+                        message: "revoke rejected: viewer still present".into(),
+                    }
+                }
             }
             network::ControlMessage::SessionCountReq { stream_id } => {
                 let count = self.local_live_sessions(&stream_id);
@@ -1081,6 +1112,24 @@ impl ClusterManager {
         }
     }
 
+    /// True when a peer `DrainStream` hint is backed by local durable state
+    /// (Raft already applied begin-delete / the row is gone).
+    fn stream_allows_network_drain(&self, stream_id: &str) -> bool {
+        match self.db.stream_get(stream_id) {
+            crate::db::DbLookup::Missing => true,
+            crate::db::DbLookup::Failed => false,
+            crate::db::DbLookup::Ok(_) => {
+                self.db.stream_pending_delete(stream_id) == Some(true)
+            }
+        }
+    }
+
+    /// True when a peer `RevokeViewer` hint targets a viewer that is no longer
+    /// in the local DB (Raft DeleteViewer already applied).
+    fn viewer_allows_network_revoke(&self, viewer_id: &str) -> bool {
+        !self.db.viewer_exists(viewer_id)
+    }
+
     /// Clone hooks out of the mutex before invoking callbacks that may need
     /// to re-enter `session_hooks` (e.g. `mark_stream_draining`).
     fn force_unpublish_and_drain(&self, stream_id: &str) {

src/cluster/network.rs

@@ -95,9 +95,15 @@ pub enum ControlMessage {
     },
     AdminDrain {
         node_id: NodeId,
+        /// HTTP-API `admin_proof` over `AdminDrain:{node_id}`.
+        #[serde(default)]
+        proof: String,
     },
     AdminResume {
         node_id: NodeId,
+        /// HTTP-API `admin_proof` over `AdminResume:{node_id}`.
+        #[serde(default)]
+        proof: String,
     },
     AdminRemove {
         node_id: NodeId,
@@ -760,11 +766,8 @@ async fn handle_control_conn<S: AsyncRead + AsyncWrite + Unpin>(
             use crate::cluster::command::ClusterCommand;
             let cmd: ClusterCommand =
                 serde_json::from_value(req.clone()).map_err(|e| std::io::Error::other(e))?;
-            if !cmd.allowed_on_control_plane() {
-                return Err(std::io::Error::other(
-                    "cluster command not allowed on control plane",
-                ));
-            }
+            // Every ClientWrite must carry an HTTP-API admin_proof (see
+            // ClusterCommand::requires_admin_proof).
             if cmd.requires_admin_proof() {
                 let req_str = serde_json::to_string(&req)
                     .map_err(|e| std::io::Error::other(e.to_string()))?;
@@ -857,6 +860,18 @@ async fn handle_control_conn<S: AsyncRead + AsyncWrite + Unpin>(
             if !is_member(peer_id) {
                 return Err(std::io::Error::other("peer not in membership"));
             }
+            if let ControlMessage::AdminDrain { node_id, proof } = &admin {
+                let payload = format!("AdminDrain:{node_id}");
+                if !verify_admin_proof(proof, &payload) {
+                    return Err(std::io::Error::other("invalid admin drain proof"));
+                }
+            }
+            if let ControlMessage::AdminResume { node_id, proof } = &admin {
+                let payload = format!("AdminResume:{node_id}");
+                if !verify_admin_proof(proof, &payload) {
+                    return Err(std::io::Error::other("invalid admin resume proof"));
+                }
+            }
             on_admin(admin)
         }
         other => {

src/db.rs

@@ -798,9 +798,24 @@ impl Db {
         .is_ok()
     }
 
-    /// Cascade: remove dependent rows so deleted streams cannot leave ghost
-    /// active publishers/players that pollute stats after stream re-creation.
-    ///
+    /// Returns `Some(true)` when `pending_delete=1`, `Some(false)` when the row
+    /// exists and is not pending, `None` when the row is missing or on DB error.
+    pub fn stream_pending_delete(&self, id: &str) -> Option<bool> {
+        let conn = self.conn.lock();
+        match conn.query_row(
+            "SELECT pending_delete FROM streams WHERE id=?",
+            params![id],
+            |r| r.get::<_, i64>(0),
+        ) {
+            Ok(v) => Some(v != 0),
+            Err(rusqlite::Error::QueryReturnedNoRows) => None,
+            Err(e) => {
+                crate::log_error!("stream_pending_delete: lookup failed for {id}: {e}");
+                None
+            }
+        }
+    }
+
     /// Like [`Self::stream_delete`], but only when `pending_delete=1`.
     /// Returns `Some(false)` when the row is missing or not pending (stale
     /// finalize must be a no-op).
@@ -1003,6 +1018,25 @@ impl Db {
         ))
     }
 
+    /// True when any viewer row with this id still exists (any stream).
+    pub fn viewer_exists(&self, viewer_id: &str) -> bool {
+        let conn = self.conn.lock();
+        match conn.query_row(
+            "SELECT 1 FROM stream_viewers WHERE id=? LIMIT 1",
+            params![viewer_id],
+            |_| Ok(()),
+        ) {
+            Ok(()) => true,
+            Err(rusqlite::Error::QueryReturnedNoRows) => false,
+            Err(e) => {
+                crate::log_error!("viewer_exists failed: {e}");
+                // Fail closed: treat as present so spoofed revokes are ignored
+                // until the DB is readable again.
+                true
+            }
+        }
+    }
+
     /// Returns `Some(true)` if deleted, `Some(false)` if not found, `None` on DB error.
     pub fn viewer_delete(&self, stream_id: &str, viewer_id: &str) -> Option<bool> {
         let conn = self.conn.lock();

src/http.rs

@@ -1210,14 +1210,19 @@ async fn finalize_stream_delete(state: &Arc<AppState>, id: &str) -> Result<(), (
 }
 
 /// How long to wait before logging that an async stream delete is taking
-/// longer than expected. The drain loop never gives up: `deleted_streams`
-/// must stay populated until every live RTMP session is gone so the poll loop
-/// keeps kicking them; removing the marker early left stuck publishers
-/// broadcasting on a disabled/pending-delete stream indefinitely.
+/// longer than expected. The drain loop keeps `deleted_streams` populated so
+/// the poll loop can kick live sessions; after [`DELETE_DRAIN_TIMEOUT`] it
+/// abandons stuck ConnState roles and finalizes anyway.
 #[cfg(test)]
 const DELETE_DRAIN_WARN_AFTER: Duration = Duration::from_millis(50);
 #[cfg(not(test))]
 const DELETE_DRAIN_WARN_AFTER: Duration = Duration::from_secs(30);
+/// Cap how long delete waits for RTMP drain. Stuck ConnState roles (failed DB
+/// deactivate) must not block finalize forever — abandon local roles then proceed.
+#[cfg(test)]
+const DELETE_DRAIN_TIMEOUT: Duration = Duration::from_millis(200);
+#[cfg(not(test))]
+const DELETE_DRAIN_TIMEOUT: Duration = Duration::from_secs(300);
 
 async fn wait_and_finalize_stream_delete(state: Arc<AppState>, id: String) {
     let started = Instant::now();
@@ -1237,6 +1242,16 @@ async fn wait_and_finalize_stream_delete(state: Arc<AppState>, id: String) {
         if local == 0 && remote == 0 {
             break;
         }
+        if started.elapsed() >= DELETE_DRAIN_TIMEOUT {
+            crate::log_error!(
+                "Delete drain timed out for stream '{id}' after {}s \
+                 (local={local} remote={remote}); abandoning local roles and finalizing",
+                DELETE_DRAIN_TIMEOUT.as_secs().max(1)
+            );
+            state.rtmp_bridge.abandon_roles_for_stream(&id);
+            state.deleted_streams.lock().insert(id.clone());
+            break;
+        }
         if last_warn.elapsed() >= DELETE_DRAIN_WARN_AFTER {
             crate::log_warn!(
                 "Still deleting stream '{id}' — stream stays disabled; \