diff --git a/all_code.txt b/all_code.txt index 81525f0..e69d7c6 100644 --- a/all_code.txt +++ b/all_code.txt @@ -768,9 +768,13 @@ def save_settings(settings_data): with open(SETTINGS_FILE, "w", encoding="utf-8") as f: json.dump(settings_data, f, indent=4, ensure_ascii=False) -# 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 -cmts_config_locks = defaultdict(asyncio.Lock) +# 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 (安全動態獲取版) +_cmts_config_locks = {} +def get_cmts_lock(host: str) -> asyncio.Lock: + if host not in _cmts_config_locks: + _cmts_config_locks[host] = asyncio.Lock() + return _cmts_config_locks[host] ================================================================================ FILE: index.html @@ -1696,7 +1700,13 @@ async def sync_cmts_leaves_async(host, username, password, leaf_paths: list): logger.info(f"🔄 正在處理第 {batch_idx + 1}/{len(batches)} 批次 (共 {len(batch)} 個路徑)...") try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=1.0, prompt_pattern: str = None): @@ -1826,7 +1836,13 @@ async def sync_cmts_leaves_async(host, username, password, leaf_paths: list): async def fetch_raw_config(host: str, username: str, password: str, config_type: str = "running") -> str: try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=2.0, prompt_pattern: str = None): @@ -2030,9 +2046,8 @@ MANAGED_LOGGERS = [ "app.scraper", # 爬蟲與快取 "app.diagnostics", # CM 診斷 "app.backup", # 備份與還原 - "app.ssh" # SSH 連線引擎 - "app.ssh" # SSH 連線引擎 - "app.auth" # 🌟 新增:上帝模式與安全授權 + "app.ssh", # 🔌 SSH 連線引擎 (注意這裡要有逗號!) + "app.auth" # 🔐 上帝模式與安全授權 ] def setup_logger(name: str, level=logging.INFO): @@ -7303,7 +7318,7 @@ from database import ( from cmts_scraper import fetch_raw_config # 🌟 [Priority 1 修復] 引入 cmts_config_locks -from shared import parse_cli_to_tree, cmts_config_locks +from shared import parse_cli_to_tree, get_cmts_lock logger = get_logger("app.backup") @@ -7522,8 +7537,9 @@ async def analyze_backup_diff(backup_id: str, req: RestoreRequest): if not raw_current: return {"status": "error", "message": "無法從設備取得當前配置,請檢查連線狀態"} - current_tree = parse_cli_to_tree(raw_current) - diff_commands = generate_diff_commands(current_tree, snapshot_tree) + # ✅ 安全升級:將 CPU 密集運算移交給 ThreadPool,保護 Event Loop 不卡死 + current_tree = await asyncio.to_thread(parse_cli_to_tree, raw_current) + diff_commands = await asyncio.to_thread(generate_diff_commands, current_tree, snapshot_tree) return { "status": "success", @@ -7549,7 +7565,7 @@ async def execute_restore(backup_id: str, req: ExecuteRestoreRequest): async def restore_generator(): # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 - host_lock = cmts_config_locks[req.host] + host_lock = get_cmts_lock[req.host] # 1. 嘗試取得全域寫入鎖 (避免多人同時打指令) if host_lock.locked(): @@ -7561,7 +7577,13 @@ async def execute_restore(backup_id: str, req: ExecuteRestoreRequest): yield json.dumps({"status": "progress", "message": "🔄 正在建立 SSH 安全連線..."}) + "\n" # 🌟 [Priority 2 提前修復] 使用 async with 確保連線與 Process 絕對會被釋放 - async with asyncssh.connect(req.host, username=req.username, password=req.password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=1.0, prompt_pattern: str = None): @@ -7672,7 +7694,10 @@ async def websocket_terminal(websocket: WebSocket, host: str, username: str, pas ssh_task = None try: # 建立連線 - conn = await asyncssh.connect(host, username=username, password=password, known_hosts=None) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) # 💡 關鍵修復:使用 term_size 參數來指定寬高 (width, height) process = await conn.create_process( @@ -7878,6 +7903,7 @@ async def get_lock_status(host: str = Query(None)): FILE: routers/diagnostics.py ================================================================================ # --- routers/diagnostics.py --- +import asyncio import re import asyncssh from logger import get_logger @@ -7913,7 +7939,14 @@ async def get_cm_diagnostics( } try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: + logger.debug(f"🚀 [Diagnostics] 開始診斷 MAC: {mac}") # ========================================== @@ -8134,7 +8167,13 @@ async def execute_single_command(host, username, password, command, timeout=15.0 自動防呆:確保指令帶有取消分頁的後綴,防止 Event Loop 卡死 """ try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: # 確保指令包含取消分頁的參數 if "| nomore" not in command.lower(): command += " | nomore" @@ -8155,7 +8194,13 @@ async def execute_interactive_command(host, username, password, commands: list, 動態處理終端機的 --More-- 提示 """ try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24)) as process: async def read_until_quiet(wait_time): @@ -8457,7 +8502,7 @@ from pydantic import BaseModel from typing import List from netmiko import ConnectHandler # 🌟 [Priority 1 修復] 引入 cmts_config_locks -from shared import CMTS_DEVICE, cmts_config_locks, parse_cli_to_tree, deep_split_tree +from shared import CMTS_DEVICE, get_cmts_lock, parse_cli_to_tree, deep_split_tree from collections import defaultdict from fastapi.responses import StreamingResponse from logger import get_all_log_levels, set_log_level @@ -8511,18 +8556,18 @@ async def execute_cmts_config(req: ConfigRequest): async def config_streamer(): # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 - host_lock = cmts_config_locks[req.host] + host_lock = get_cmts_lock[req.host] async with host_lock: try: yield json.dumps({"status": "progress", "message": f"🔌 準備連線至設備 {req.host}..."}) + "\n" - # 使用 asyncssh 建立非同步連線 - async with asyncssh.connect( - req.host, - username=req.username, - password=req.password, - known_hosts=None - ) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process() as process: # 🌟 建立安全的非同步讀取函數 (讀到安靜為止,避免卡死) @@ -8938,6 +8983,9 @@ async def sse_stream(request: Request, host: str): finally: if channel_key in active_clients: active_clients[channel_key].discard(client_queue) + # ✅ 安全升級:如果該設備已經沒有任何客戶端監聽,徹底刪除 Key 釋放記憶體 + if not active_clients[channel_key]: + del active_clients[channel_key] return StreamingResponse(event_generator(), media_type="text/event-stream") diff --git a/cmts_scraper.py b/cmts_scraper.py index 6950e96..747091d 100644 --- a/cmts_scraper.py +++ b/cmts_scraper.py @@ -154,7 +154,13 @@ async def sync_cmts_leaves_async(host, username, password, leaf_paths: list): logger.info(f"🔄 正在處理第 {batch_idx + 1}/{len(batches)} 批次 (共 {len(batch)} 個路徑)...") try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=1.0, prompt_pattern: str = None): @@ -284,7 +290,13 @@ async def sync_cmts_leaves_async(host, username, password, leaf_paths: list): async def fetch_raw_config(host: str, username: str, password: str, config_type: str = "running") -> str: try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=2.0, prompt_pattern: str = None): diff --git a/logger.py b/logger.py index e972326..56a2ae8 100644 --- a/logger.py +++ b/logger.py @@ -26,9 +26,8 @@ MANAGED_LOGGERS = [ "app.scraper", # 爬蟲與快取 "app.diagnostics", # CM 診斷 "app.backup", # 備份與還原 - "app.ssh" # SSH 連線引擎 - "app.ssh" # SSH 連線引擎 - "app.auth" # 🌟 新增:上帝模式與安全授權 + "app.ssh", # 🔌 SSH 連線引擎 (注意這裡要有逗號!) + "app.auth" # 🔐 上帝模式與安全授權 ] def setup_logger(name: str, level=logging.INFO): diff --git a/routers/backup.py b/routers/backup.py index cb44c47..4173772 100644 --- a/routers/backup.py +++ b/routers/backup.py @@ -23,7 +23,7 @@ from database import ( from cmts_scraper import fetch_raw_config # 🌟 [Priority 1 修復] 引入 cmts_config_locks -from shared import parse_cli_to_tree, cmts_config_locks +from shared import parse_cli_to_tree, get_cmts_lock logger = get_logger("app.backup") @@ -242,8 +242,9 @@ async def analyze_backup_diff(backup_id: str, req: RestoreRequest): if not raw_current: return {"status": "error", "message": "無法從設備取得當前配置,請檢查連線狀態"} - current_tree = parse_cli_to_tree(raw_current) - diff_commands = generate_diff_commands(current_tree, snapshot_tree) + # ✅ 安全升級:將 CPU 密集運算移交給 ThreadPool,保護 Event Loop 不卡死 + current_tree = await asyncio.to_thread(parse_cli_to_tree, raw_current) + diff_commands = await asyncio.to_thread(generate_diff_commands, current_tree, snapshot_tree) return { "status": "success", @@ -269,7 +270,7 @@ async def execute_restore(backup_id: str, req: ExecuteRestoreRequest): async def restore_generator(): # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 - host_lock = cmts_config_locks[req.host] + host_lock = get_cmts_lock[req.host] # 1. 嘗試取得全域寫入鎖 (避免多人同時打指令) if host_lock.locked(): @@ -281,7 +282,13 @@ async def execute_restore(backup_id: str, req: ExecuteRestoreRequest): yield json.dumps({"status": "progress", "message": "🔄 正在建立 SSH 安全連線..."}) + "\n" # 🌟 [Priority 2 提前修復] 使用 async with 確保連線與 Process 絕對會被釋放 - async with asyncssh.connect(req.host, username=req.username, password=req.password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process: async def read_until_quiet(timeout=1.0, prompt_pattern: str = None): diff --git a/routers/config.py b/routers/config.py index 3ddab08..2adac35 100644 --- a/routers/config.py +++ b/routers/config.py @@ -10,7 +10,7 @@ from pydantic import BaseModel from typing import List from netmiko import ConnectHandler # 🌟 [Priority 1 修復] 引入 cmts_config_locks -from shared import CMTS_DEVICE, cmts_config_locks, parse_cli_to_tree, deep_split_tree +from shared import CMTS_DEVICE, get_cmts_lock, parse_cli_to_tree, deep_split_tree from collections import defaultdict from fastapi.responses import StreamingResponse from logger import get_all_log_levels, set_log_level @@ -64,18 +64,18 @@ async def execute_cmts_config(req: ConfigRequest): async def config_streamer(): # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 - host_lock = cmts_config_locks[req.host] + host_lock = get_cmts_lock[req.host] async with host_lock: try: yield json.dumps({"status": "progress", "message": f"🔌 準備連線至設備 {req.host}..."}) + "\n" - # 使用 asyncssh 建立非同步連線 - async with asyncssh.connect( - req.host, - username=req.username, - password=req.password, - known_hosts=None - ) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process() as process: # 🌟 建立安全的非同步讀取函數 (讀到安靜為止,避免卡死) diff --git a/routers/diagnostics.py b/routers/diagnostics.py index b877582..3651225 100644 --- a/routers/diagnostics.py +++ b/routers/diagnostics.py @@ -1,4 +1,5 @@ # --- routers/diagnostics.py --- +import asyncio import re import asyncssh from logger import get_logger @@ -34,7 +35,14 @@ async def get_cm_diagnostics( } try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: + logger.debug(f"🚀 [Diagnostics] 開始診斷 MAC: {mac}") # ========================================== diff --git a/routers/leaf_options.py b/routers/leaf_options.py index e1f25ff..060e3f5 100644 --- a/routers/leaf_options.py +++ b/routers/leaf_options.py @@ -50,6 +50,9 @@ async def sse_stream(request: Request, host: str): finally: if channel_key in active_clients: active_clients[channel_key].discard(client_queue) + # ✅ 安全升級:如果該設備已經沒有任何客戶端監聽,徹底刪除 Key 釋放記憶體 + if not active_clients[channel_key]: + del active_clients[channel_key] return StreamingResponse(event_generator(), media_type="text/event-stream") diff --git a/routers/query.py b/routers/query.py index 1703f04..521f3e9 100644 --- a/routers/query.py +++ b/routers/query.py @@ -17,7 +17,13 @@ async def execute_single_command(host, username, password, command, timeout=15.0 自動防呆:確保指令帶有取消分頁的後綴,防止 Event Loop 卡死 """ try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: # 確保指令包含取消分頁的參數 if "| nomore" not in command.lower(): command += " | nomore" @@ -38,7 +44,13 @@ async def execute_interactive_command(host, username, password, commands: list, 動態處理終端機的 --More-- 提示 """ try: - async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn: + # 先用 wait_for 取得連線 (超過 10 秒會拋出 asyncio.TimeoutError) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) + # 再進入 async with 確保資源會被自動關閉 + async with conn: async with conn.create_process(term_type='xterm-256color', term_size=(200, 24)) as process: async def read_until_quiet(wait_time): diff --git a/routers/terminal.py b/routers/terminal.py index 8ad4480..9c29e7d 100644 --- a/routers/terminal.py +++ b/routers/terminal.py @@ -15,7 +15,10 @@ async def websocket_terminal(websocket: WebSocket, host: str, username: str, pas ssh_task = None try: # 建立連線 - conn = await asyncssh.connect(host, username=username, password=password, known_hosts=None) + conn = await asyncio.wait_for( + asyncssh.connect(host, username=username, password=password, known_hosts=None), + timeout=10.0 + ) # 💡 關鍵修復:使用 term_size 參數來指定寬高 (width, height) process = await conn.create_process( diff --git a/shared.py b/shared.py index cc4fe60..436d8a6 100644 --- a/shared.py +++ b/shared.py @@ -257,5 +257,10 @@ def save_settings(settings_data): with open(SETTINGS_FILE, "w", encoding="utf-8") as f: json.dump(settings_data, f, indent=4, ensure_ascii=False) -# 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 -cmts_config_locks = defaultdict(asyncio.Lock) +# 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 (安全動態獲取版) +_cmts_config_locks = {} + +def get_cmts_lock(host: str) -> asyncio.Lock: + if host not in _cmts_config_locks: + _cmts_config_locks[host] = asyncio.Lock() + return _cmts_config_locks[host] \ No newline at end of file