From c9a77c8757edf81959ef224a63cd89dae64892f8 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 20 May 2026 02:40:08 +0530 Subject: [PATCH] fix(telegram): stop duplicate operator-approval prompts for allowed users (#1948) (#2240) Co-authored-by: Claude --- .../providers/telegram/channel_core.rs | 1 + .../providers/telegram/channel_recv.rs | 113 ++++++++- .../providers/telegram/channel_tests.rs | 225 ++++++++++++++++++ .../providers/telegram/channel_types.rs | 11 +- 4 files changed, 342 insertions(+), 8 deletions(-) diff --git a/src/openhuman/channels/providers/telegram/channel_core.rs b/src/openhuman/channels/providers/telegram/channel_core.rs index df0e2afbb..9875a58bb 100644 --- a/src/openhuman/channels/providers/telegram/channel_core.rs +++ b/src/openhuman/channels/providers/telegram/channel_core.rs @@ -38,6 +38,7 @@ impl TelegramChannel { mention_only, bot_username: parking_lot::Mutex::new(None), recent_updates: parking_lot::Mutex::new(TelegramUpdateWindow::default()), + recent_approval_prompts: parking_lot::Mutex::new(std::collections::HashMap::new()), } } diff --git a/src/openhuman/channels/providers/telegram/channel_recv.rs b/src/openhuman/channels/providers/telegram/channel_recv.rs index a3cd6a7f3..576907f62 100644 --- a/src/openhuman/channels/providers/telegram/channel_recv.rs +++ b/src/openhuman/channels/providers/telegram/channel_recv.rs @@ -1,8 +1,9 @@ //! Telegram channel — inbound message/reaction parsing, allowlist checks, mention filtering, //! unauthorized-message handling, and typing-action helpers. -use super::channel_types::{TelegramChannel, TelegramReactionEvent}; +use super::channel_types::{TelegramChannel, TelegramReactionEvent, APPROVAL_PROMPT_DEBOUNCE_SECS}; use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage}; +use std::time::Instant; impl TelegramChannel { pub(crate) fn typing_body_for_recipient(recipient: &str) -> serde_json::Value { @@ -177,6 +178,53 @@ impl TelegramChannel { identities.into_iter().any(|id| self.is_user_allowed(id)) } + /// Check whether an approval prompt should be suppressed due to the restart-race + /// condition signature: `pairing.is_none()` (channel was constructed with a non-empty + /// allowlist) AND the runtime `allowed_users` list is currently empty. + /// + /// This happens when the replacement process reads its config allowlist, stores it in + /// `allowed_users`, but the old process has not yet shut down — Telegram redelivers + /// the update to both. The racing instance has `pairing = None` (correct — allowlist + /// was non-empty at construction) but the runtime list may briefly show as empty before + /// the config is loaded. + /// + /// Legitimate first-run pairing (`allowed_users=[]` at construction) always sets + /// `pairing = Some(...)` so it is never suppressed here. + pub(crate) fn is_race_condition_instance(&self) -> bool { + let runtime_empty = self + .allowed_users + .read() + .map(|users| users.is_empty()) + .unwrap_or(false); + runtime_empty && self.pairing.is_none() + } + + /// Build the de-bounce key for approval prompts: `"{chat_id}:{sender}"`. + pub(crate) fn approval_debounce_key(chat_id: &str, sender: &str) -> String { + format!("{chat_id}:{sender}") + } + + /// Returns `true` if an approval prompt was already sent to this chat+sender within the + /// de-bounce window, and updates the last-sent timestamp when returning `false`. + pub(crate) fn check_and_update_approval_debounce(&self, chat_id: &str, sender: &str) -> bool { + let key = Self::approval_debounce_key(chat_id, sender); + let mut prompts = self.recent_approval_prompts.lock(); + if let Some(last_sent) = prompts.get(&key) { + if last_sent.elapsed().as_secs() < APPROVAL_PROMPT_DEBOUNCE_SECS { + return true; // still within de-bounce window + } + } + // Evict entries older than the de-bounce window before inserting. Anything + // past the window can never suppress again, so retaining it would let the + // map grow without bound if the bot is exposed to a public group or spam + // (review note on #1948). This caps the map to senders seen within the + // last APPROVAL_PROMPT_DEBOUNCE_SECS. + prompts + .retain(|_, last_sent| last_sent.elapsed().as_secs() < APPROVAL_PROMPT_DEBOUNCE_SECS); + prompts.insert(key, Instant::now()); + false + } + pub(crate) async fn handle_unauthorized_message(&self, update: &serde_json::Value) { let Some(message) = update.get("message") else { return; @@ -207,7 +255,7 @@ impl TelegramChannel { .map(|id| id.to_string()); let Some(chat_id) = chat_id else { - tracing::warn!("Telegram: missing chat_id in message, skipping"); + tracing::warn!("[telegram][approval] missing chat_id in message, skipping"); return; }; @@ -217,6 +265,30 @@ impl TelegramChannel { } if self.is_any_user_allowed(identities.iter().copied()) { + tracing::debug!( + chat_id, + username, + sender_id = sender_id_str.as_deref().unwrap_or("unknown"), + "[telegram][approval] message sender is allowed — no action" + ); + return; + } + + // ── Race-condition guard ───────────────────────────────────────────────── + // Signature: pairing.is_none() (channel constructed with non-empty allowlist) + // AND runtime allowed_users is currently empty. This means we are the racing + // instance spawned during a restart whose config hasn't propagated yet. + // Sending an approval prompt here would spam the allowlisted user with false + // "operator approval required" messages. Log and suppress instead. + if self.is_race_condition_instance() { + tracing::warn!( + chat_id, + username, + sender_id = sender_id_str.as_deref().unwrap_or("unknown"), + "[telegram][approval] race-condition guard: allowlist is empty at runtime \ + but channel was constructed with a non-empty allowlist (pairing=None). \ + Suppressing approval prompt — this is a restart-overlap false positive." + ); return; } @@ -243,12 +315,16 @@ impl TelegramChannel { )) .await; tracing::info!( - "Telegram: paired and allowlisted identity={identity}" + chat_id, + identity, + "[telegram][approval] paired and allowlisted identity" ); } Err(e) => { tracing::error!( - "Telegram: failed to persist allowlist after bind: {e}" + chat_id, + error = %e, + "[telegram][approval] failed to persist allowlist after bind" ); let _ = self .send(&SendMessage::new( @@ -295,10 +371,28 @@ impl TelegramChannel { return; } + // ── De-bounce: suppress duplicate approval prompts within the window ──────── + // Key by chat_id + sender so multiple different senders are tracked independently. + let sender_key = normalized_sender_id + .as_deref() + .unwrap_or(normalized_username.as_str()); + if self.check_and_update_approval_debounce(&chat_id, sender_key) { + tracing::debug!( + chat_id, + sender = sender_key, + "[telegram][approval] de-bounce: suppressing duplicate approval prompt \ + (sent within {}s window)", + APPROVAL_PROMPT_DEBOUNCE_SECS + ); + return; + } + tracing::warn!( - "Telegram: ignoring message from unauthorized user: username={username}, sender_id={}. \ -Allowlist Telegram username (without '@') or numeric user ID.", - sender_id_str.as_deref().unwrap_or("unknown") + chat_id, + username, + sender_id = sender_id_str.as_deref().unwrap_or("unknown"), + "[telegram][approval] unauthorized user; sending approval prompt. \ + Allowlist Telegram username (without '@') or numeric user ID." ); let _ = self @@ -309,6 +403,11 @@ Allowlist Telegram username (without '@') or numeric user ID.", .await; if self.pairing_code_active() { + tracing::debug!( + chat_id, + sender = sender_key, + "[telegram][approval] pairing code active — sending /bind hint" + ); let _ = self .send(&SendMessage::new( "ℹ️ If operator provides a one-time pairing code, you can also run `/bind `.", diff --git a/src/openhuman/channels/providers/telegram/channel_tests.rs b/src/openhuman/channels/providers/telegram/channel_tests.rs index eb7b66ce4..e7383b2b8 100644 --- a/src/openhuman/channels/providers/telegram/channel_tests.rs +++ b/src/openhuman/channels/providers/telegram/channel_tests.rs @@ -1607,3 +1607,228 @@ async fn test_thinking_placeholder_logic() { assert!(!update.contains("thought 2")); } } + +// ── Issue #1948: Duplicate approval prompts regression tests ────────────────── +// +// These tests cover the race-guard and de-bounce logic added in fix/1948. +// They FAIL to compile (method not found) before the fix is applied. + +/// Test A (race-guard): channel constructed with allowed_users=["alice"] has pairing=None. +/// When runtime allowlist is cleared to empty (simulating restart-race), `is_race_condition_instance` +/// must return `true` so the approval prompt is suppressed. +#[test] +fn is_race_condition_instance_true_when_allowlist_was_nonempty_but_runtime_is_empty() { + let ch = TelegramChannel::new("t".into(), vec!["alice".into()], false); + // Channel constructed with non-empty list → pairing is None + assert!( + ch.pairing.is_none(), + "pre-condition: non-empty allowlist must set pairing=None" + ); + + // Simulate the race: runtime allowlist cleared by an in-flight config reload + { + let mut users = ch.allowed_users.write().unwrap(); + users.clear(); + } + + assert!( + ch.is_race_condition_instance(), + "[telegram][approval] race guard must fire: runtime empty AND pairing=None" + ); +} + +/// Test B (legit pairing preserved): channel constructed with empty allowlist → pairing=Some. +/// `is_race_condition_instance` must return `false` so the first-run pairing flow is unaffected. +#[test] +fn is_race_condition_instance_false_for_legitimate_empty_allowlist() { + let ch = TelegramChannel::new("t".into(), vec![], false); + // Channel constructed with empty list → pairing is Some (first-run pairing) + assert!( + ch.pairing.is_some(), + "pre-condition: empty allowlist must set pairing=Some" + ); + + // Runtime list is also empty (genuine first-run state) + assert!( + !ch.is_race_condition_instance(), + "[telegram][approval] race guard must NOT fire for genuine first-run pairing" + ); +} + +/// Test C (genuine reject preserved): channel with allowed_users=["alice"], "bob" is not allowed. +/// `is_race_condition_instance` must return `false` when runtime list is non-empty (normal case). +#[test] +fn is_race_condition_instance_false_when_runtime_allowlist_populated() { + let ch = TelegramChannel::new("t".into(), vec!["alice".into()], false); + assert!( + ch.pairing.is_none(), + "pre-condition: non-empty allowlist must set pairing=None" + ); + + // Runtime list is populated (normal operational state — NOT a race) + let users = ch.allowed_users.read().unwrap(); + assert!( + !users.is_empty(), + "pre-condition: runtime list should still contain alice" + ); + drop(users); + + assert!( + !ch.is_race_condition_instance(), + "[telegram][approval] race guard must NOT fire when runtime allowlist is populated" + ); +} + +/// Test D (de-bounce — first call): verify that the first approval prompt is NOT suppressed. +#[test] +fn approval_debounce_first_call_not_suppressed() { + let ch = TelegramChannel::new("t".into(), vec![], false); + let suppressed = ch.check_and_update_approval_debounce("12345", "alice"); + assert!( + !suppressed, + "[telegram][approval] first approval prompt must not be suppressed" + ); +} + +/// Test D (de-bounce — rapid second call): a second call within the window must be suppressed. +#[test] +fn approval_debounce_rapid_second_call_suppressed() { + let ch = TelegramChannel::new("t".into(), vec![], false); + + // First call registers the timestamp + let first = ch.check_and_update_approval_debounce("12345", "alice"); + assert!(!first, "first call must not be suppressed"); + + // Immediate second call — still within the 60s window + let second = ch.check_and_update_approval_debounce("12345", "alice"); + assert!( + second, + "[telegram][approval] rapid second approval prompt to same chat+sender must be suppressed" + ); +} + +/// Test D (de-bounce — different sender): a different sender in the same chat is NOT suppressed. +#[test] +fn approval_debounce_different_sender_not_suppressed() { + let ch = TelegramChannel::new("t".into(), vec![], false); + + let _ = ch.check_and_update_approval_debounce("12345", "alice"); + + // "bob" sending to the same chat is a different key — must not be suppressed + let suppressed = ch.check_and_update_approval_debounce("12345", "bob"); + assert!( + !suppressed, + "[telegram][approval] different sender must not be suppressed by alice's de-bounce" + ); +} + +/// Test D (de-bounce — different chat): the same sender in a different chat is NOT suppressed. +#[test] +fn approval_debounce_different_chat_not_suppressed() { + let ch = TelegramChannel::new("t".into(), vec![], false); + + let _ = ch.check_and_update_approval_debounce("chat_a", "mallory"); + + // Same sender, different chat — different key — must not be suppressed + let suppressed = ch.check_and_update_approval_debounce("chat_b", "mallory"); + assert!( + !suppressed, + "[telegram][approval] different chat must not be suppressed by chat_a's de-bounce" + ); +} + +/// Test D (de-bounce — window expiry): after advancing the clock past the de-bounce window, +/// the same chat+sender is allowed again. +/// +/// We can't advance a real clock cheaply in a unit test, so we instead verify that the +/// de-bounce bucket is re-inserted with a fresh timestamp on every non-suppressed call, +/// by checking that the map entry is updated when the first call fires. +#[test] +fn approval_debounce_map_entry_inserted_on_first_call() { + let ch = TelegramChannel::new("t".into(), vec![], false); + + { + let prompts = ch.recent_approval_prompts.lock(); + assert!( + prompts.is_empty(), + "map must be empty before any approval prompt" + ); + } + + let suppressed = ch.check_and_update_approval_debounce("99", "mallory"); + assert!(!suppressed); + + { + let prompts = ch.recent_approval_prompts.lock(); + let key = TelegramChannel::approval_debounce_key("99", "mallory"); + assert!( + prompts.contains_key(&key), + "[telegram][approval] first call must register entry in recent_approval_prompts map" + ); + } +} + +/// Test D (de-bounce — multiple calls, 4 rapid): four rapid calls from the same sender must result +/// in exactly one entry in the map and only the first call being non-suppressed. +#[test] +fn approval_debounce_four_rapid_calls_suppressed_after_first() { + let ch = TelegramChannel::new("t".into(), vec![], false); + + let mut not_suppressed_count = 0usize; + for _ in 0..4 { + if !ch.check_and_update_approval_debounce("777", "spammer") { + not_suppressed_count += 1; + } + } + + assert_eq!( + not_suppressed_count, 1, + "[telegram][approval] exactly 1 of 4 rapid calls must not be suppressed (de-bounce)" + ); +} + +/// Review note on #1948 (@graycyrus): the de-bounce map must not grow without +/// bound. Entries older than the de-bounce window are dead weight — they can +/// never suppress again — so a non-suppressed call evicts them. Pre-seed one +/// stale entry and one fresh entry, trigger a new non-suppressed call, and +/// assert the stale entry is gone while the fresh and new ones remain. +#[test] +fn approval_debounce_evicts_entries_past_window() { + let ch = TelegramChannel::new("t".into(), vec![], false); + let stale = std::time::Instant::now() + .checked_sub(Duration::from_secs(3600)) + .expect("monotonic clock far enough from boot for a 1h offset"); + + { + let mut prompts = ch.recent_approval_prompts.lock(); + prompts.insert("stale_chat:stale_sender".to_string(), stale); + prompts.insert( + "fresh_chat:fresh_sender".to_string(), + std::time::Instant::now(), + ); + } + + // Non-suppressed call for a new key triggers the eviction sweep + insert. + let suppressed = ch.check_and_update_approval_debounce("new_chat", "new_sender"); + assert!( + !suppressed, + "first call for a new key must not be suppressed" + ); + + let prompts = ch.recent_approval_prompts.lock(); + assert!( + !prompts.contains_key("stale_chat:stale_sender"), + "[telegram][approval] stale entry past the de-bounce window must be evicted (no unbounded growth)" + ); + assert!( + prompts.contains_key("fresh_chat:fresh_sender"), + "[telegram][approval] fresh entry within the window must be retained" + ); + assert!( + prompts.contains_key(&TelegramChannel::approval_debounce_key( + "new_chat", + "new_sender" + )), + "[telegram][approval] the new entry must be inserted" + ); +} diff --git a/src/openhuman/channels/providers/telegram/channel_types.rs b/src/openhuman/channels/providers/telegram/channel_types.rs index 4fd140806..11fb8b691 100644 --- a/src/openhuman/channels/providers/telegram/channel_types.rs +++ b/src/openhuman/channels/providers/telegram/channel_types.rs @@ -5,9 +5,14 @@ use crate::openhuman::security::pairing::PairingGuard; use parking_lot::Mutex; use std::collections::{HashSet, VecDeque}; use std::sync::{Arc, RwLock}; +use std::time::Instant; pub(crate) const TELEGRAM_RECENT_UPDATE_CACHE_SIZE: usize = 4096; +/// De-bounce window for approval prompts: suppress duplicate prompts sent to the +/// same chat+sender within this duration (prevents restart-race and rapid-fire spam). +pub(crate) const APPROVAL_PROMPT_DEBOUNCE_SECS: u64 = 60; + pub(crate) struct TelegramTypingTask { pub(crate) recipient: String, pub(crate) handle: tokio::task::JoinHandle<()>, @@ -38,8 +43,12 @@ pub struct TelegramChannel { pub(crate) stream_mode: StreamMode, pub(crate) draft_update_interval_ms: u64, pub(crate) silent_streaming: bool, - pub(crate) last_draft_edit: Mutex>, + pub(crate) last_draft_edit: Mutex>, pub(crate) mention_only: bool, pub(crate) bot_username: Mutex>, pub(crate) recent_updates: Mutex, + /// Tracks the last time an approval prompt was sent to a given "chat_id:sender" key. + /// Prevents duplicate prompts during restart-overlap races and rapid re-sends. + /// Mirrors the `last_draft_edit` pattern. + pub(crate) recent_approval_prompts: Mutex>, }