Restrict global system memory viewing and tabs strictly to admin users

telegram-agy-bot/bot.py

@@ -1115,6 +1115,10 @@ def build_memory_menu(chat_id: int, view_type: str = "project", page: int = 0) -
 
     proj_name = curr_proj.name if curr_proj else "default"
 
+    # Non-admin users are strictly forbidden from viewing global memory
+    if view_type == "global" and not is_admin_user:
+        view_type = "project" if curr_proj else "user"
+
     # Validate view_type
     if view_type not in ("project", "user", "global"):
         view_type = "project" if curr_proj else "user"
@@ -1123,8 +1127,11 @@ def build_memory_menu(chat_id: int, view_type: str = "project", page: int = 0) -
         memories = memory_manager.get_project_memories(user_id=chat_id, project_name=proj_name, limit=100)
     elif view_type == "user":
         memories = memory_manager.get_user_memories(user_id=chat_id, limit=100)
-    else:  # global
+    elif view_type == "global" and is_admin_user:
         memories = memory_manager.get_global_memories(limit=100)
+    else:
+        memories = memory_manager.get_user_memories(user_id=chat_id, limit=100)
+        view_type = "user"
 
     total_cnt = len(memories)
     page_size = 5
@@ -1145,11 +1152,15 @@ def build_memory_menu(chat_id: int, view_type: str = "project", page: int = 0) -
             tab_name = "🌐 قوانین و خاطرات عمومی (کل سرور و کاربران)"
             tab_desc = "این قوانین برای <b>همه کاربران</b> تزریق می‌شوند (ویرایش و حذف فقط توسط مدیر)."
 
+        stats_str = f"<code>{stats['current_project_memories']}</code> پروژه | <code>{stats['current_user_memories']}</code> کاربر"
+        if is_admin_user:
+            stats_str += f" | <code>{stats['global_memories']}</code> عمومی"
+
         header = (
-            f"🧠 <b>کنترل‌پنل حافظه هوشمند ۳ لایه (AI Memory Dashboard)</b>\n\n"
+            f"🧠 <b>کنترل‌پنل حافظه هوشمند (AI Memory Dashboard)</b>\n\n"
             f"• 📂 <b>بخش فعال:</b> {tab_name}\n"
             f"• ℹ️ {tab_desc}\n"
-            f"• 📊 <b>آمار:</b> <code>{stats['current_project_memories']}</code> پروژه | <code>{stats['current_user_memories']}</code> کاربر | <code>{stats['global_memories']}</code> عمومی\n\n"
+            f"• 📊 <b>آمار:</b> {stats_str}\n\n"
         )
         if not memories:
             body = (
@@ -1176,11 +1187,15 @@ def build_memory_menu(chat_id: int, view_type: str = "project", page: int = 0) -
             tab_name = "🌐 Global System Memory"
             tab_desc = "Applies to all users (admin-managed)."
 
+        stats_str = f"<code>{stats['current_project_memories']}</code> project | <code>{stats['current_user_memories']}</code> user"
+        if is_admin_user:
+            stats_str += f" | <code>{stats['global_memories']}</code> global"
+
         header = (
-            f"🧠 <b>3-Tier AI Memory Dashboard</b>\n\n"
+            f"🧠 <b>AI Memory Dashboard</b>\n\n"
             f"• 📂 <b>Active Tab:</b> {tab_name}\n"
             f"• ℹ️ {tab_desc}\n"
-            f"• 📊 <b>Stats:</b> <code>{stats['current_project_memories']}</code> project | <code>{stats['current_user_memories']}</code> user | <code>{stats['global_memories']}</code> global\n\n"
+            f"• 📊 <b>Stats:</b> {stats_str}\n\n"
         )
         if not memories:
             body = "ℹ️ <i>No memories recorded in this section yet.</i>"
@@ -1199,7 +1214,7 @@ def build_memory_menu(chat_id: int, view_type: str = "project", page: int = 0) -
     # Build keyboard
     keyboard = []
 
-    # 1. 3 Tab buttons
+    # 1. Tab buttons (Global tab only visible to admin)
     t_proj = f"📁 پروژه ({proj_name[:10]}) ✓" if view_type == "project" else f"📁 پروژه ({proj_name[:10]})"
     t_user = "👤 کاربر (من) ✓" if view_type == "user" else "👤 کاربر (من)"
     t_glob = "🌐 عمومی ✓" if view_type == "global" else "🌐 عمومی"
@@ -1208,11 +1223,13 @@ def build_memory_menu(chat_id: int, view_type: str = "project", page: int = 0) -
         t_user = "👤 User ✓" if view_type == "user" else "👤 User"
         t_glob = "🌐 Global ✓" if view_type == "global" else "🌐 Global"
 
-    keyboard.append([
+    tab_row = [
         InlineKeyboardButton(t_proj, callback_data="mem_tab:project:0"),
         InlineKeyboardButton(t_user, callback_data="mem_tab:user:0"),
-        InlineKeyboardButton(t_glob, callback_data="mem_tab:global:0"),
-    ])
+    ]
+    if is_admin_user:
+        tab_row.append(InlineKeyboardButton(t_glob, callback_data="mem_tab:global:0"))
+    keyboard.append(tab_row)
 
     # 2. Item delete buttons
     can_delete_current = (view_type in ("project", "user")) or (view_type == "global" and is_admin_user)
@@ -3438,6 +3455,13 @@ async def memory_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
         return
 
     if subcmd in ("global", "g", "public", "system"):
+        if not is_admin_user:
+            await update.message.reply_html(
+                "🚫 <b>خطای دسترسی: مشاهده و مدیریت حافظه عمومی سیستم فقط مختص مدیر است.</b>"
+                if is_fa
+                else "🚫 <b>Access Denied: Viewing global system memory is restricted to administrators.</b>"
+            )
+            return
         text, markup = build_memory_menu(chat_id, view_type="global", page=0)
         await update.message.reply_html(text, reply_markup=markup)
         return
@@ -5197,8 +5221,12 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
 
     elif data == "btn_memory_menu" or data.startswith("mem_tab:"):
         parts = data.split(":")
-        v_type = parts[1] if len(parts) > 1 else "private"
+        v_type = parts[1] if len(parts) > 1 else "project"
         page_num = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0
+        if v_type == "global" and not is_admin_user:
+            await query.answer("🚫 مشاهده حافظه عمومی فقط برای مدیر سیستم مجاز است." if is_fa else "🚫 Global memory access restricted to administrators.", show_alert=True)
+            v_type = "project" if session_manager.get_current_project(chat_id) else "user"
+            page_num = 0
         text, markup = build_memory_menu(chat_id, view_type=v_type, page=page_num)
         try:
             await query.edit_message_text(text, parse_mode=constants.ParseMode.HTML, reply_markup=markup)
@@ -5215,6 +5243,9 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
             await query.answer("❌ خاطره یافت نشد." if is_fa else "❌ Memory not found.", show_alert=True)
             text, markup = build_memory_menu(chat_id, view_type=v_type, page=p_num)
             await query.edit_message_text(text, parse_mode=constants.ParseMode.HTML, reply_markup=markup)
+        elif mem.is_global and not is_admin_user:
+            await query.answer("🚫 دسترسی غیرمجاز!" if is_fa else "🚫 Permission denied!", show_alert=True)
+            return
         else:
             confirm_text = (
                 f"🗑️ <b>آیا از حذف این خاطره اطمینان دارید؟</b>\n\n"
@@ -5239,6 +5270,10 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
         m_id = int(m_id_str)
         p_num = int(page_str)
         from memory_manager import memory_manager
+        mem = memory_manager.get_by_id(m_id)
+        if mem and mem.is_global and not is_admin_user:
+            await query.answer("🚫 دسترسی غیرمجاز!" if is_fa else "🚫 Permission denied!", show_alert=True)
+            return
         try:
             ok = memory_manager.delete_by_id(m_id, user_id=chat_id, is_admin=is_admin_user)
             if ok:
@@ -5258,6 +5293,9 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
 
     elif data.startswith("mem_clear_conf:"):
         v_type = data.split(":")[1]
+        if v_type == "global" and not is_admin_user:
+            await query.answer("🚫 دسترسی غیرمجاز!" if is_fa else "🚫 Permission denied!", show_alert=True)
+            return
         type_label = "عمومی سیستم" if v_type == "global" else "اختصاصی خودتان"
         confirm_text = (
             f"⚠️ <b>هشدار پاک‌سازی حافظه</b>\n\n"
@@ -5277,6 +5315,9 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
 
     elif data.startswith("mem_clear_do:"):
         v_type = data.split(":")[1]
+        if v_type == "global" and not is_admin_user:
+            await query.answer("🚫 دسترسی غیرمجاز!" if is_fa else "🚫 Permission denied!", show_alert=True)
+            return
         from memory_manager import memory_manager
         curr_p = session_manager.get_current_project(chat_id)
         p_name = curr_p.name if curr_p else "default"

telegram-agy-bot/memory_manager.py

@@ -421,10 +421,13 @@ class MemoryManager:
         project_name: Optional[str] = None,
         type_: Optional[str] = None,
         limit: int = 50,
+        is_admin: bool = False,
     ) -> List[MemoryItem]:
         q = f"%{query.strip().lower()}%"
         with self._get_connection() as conn:
             if type_ == "global":
+                if not is_admin:
+                    return []
                 cur = conn.execute(
                     """
                     SELECT * FROM memories 
@@ -452,16 +455,29 @@ class MemoryManager:
                     (user_id, project_name, q, q, q, limit),
                 )
             elif user_id is not None:
-                cur = conn.execute(
-                    """
-                    SELECT * FROM memories 
-                    WHERE (type = 'global' OR (type IN ('user', 'private') AND user_id = ?) OR (type = 'project' AND user_id = ? AND project_name = ?)) 
-                      AND (key LIKE ? OR content LIKE ? OR category LIKE ?)
-                    ORDER BY importance DESC, updated_at DESC LIMIT ?
-                    """,
-                    (user_id, user_id, project_name or "", q, q, q, limit),
-                )
+                if is_admin:
+                    cur = conn.execute(
+                        """
+                        SELECT * FROM memories 
+                        WHERE (type = 'global' OR (type IN ('user', 'private') AND user_id = ?) OR (type = 'project' AND user_id = ? AND project_name = ?)) 
+                          AND (key LIKE ? OR content LIKE ? OR category LIKE ?)
+                        ORDER BY importance DESC, updated_at DESC LIMIT ?
+                        """,
+                        (user_id, user_id, project_name or "", q, q, q, limit),
+                    )
+                else:
+                    cur = conn.execute(
+                        """
+                        SELECT * FROM memories 
+                        WHERE ((type IN ('user', 'private') AND user_id = ?) OR (type = 'project' AND user_id = ? AND project_name = ?)) 
+                          AND (key LIKE ? OR content LIKE ? OR category LIKE ?)
+                        ORDER BY importance DESC, updated_at DESC LIMIT ?
+                        """,
+                        (user_id, user_id, project_name or "", q, q, q, limit),
+                    )
             else:
+                if not is_admin:
+                    return []
                 cur = conn.execute(
                     """
                     SELECT * FROM memories