diff --git a/routers/backup.py b/routers/backup.py index c461e32..b689b0b 100644 --- a/routers/backup.py +++ b/routers/backup.py @@ -1,3 +1,4 @@ +# --- routers/backup.py --- import logging import asyncssh import asyncio @@ -17,7 +18,8 @@ from database import ( ) from cmts_scraper import fetch_raw_config -from shared import parse_cli_to_tree, cmts_config_lock # <== 引入真正的解析器與全域鎖 +# 🌟 [Priority 1 修復] 引入 cmts_config_locks +from shared import parse_cli_to_tree, cmts_config_locks logger = logging.getLogger(__name__) @@ -223,97 +225,101 @@ async def execute_restore(backup_id: str, req: ExecuteRestoreRequest): return StreamingResponse(empty_success(), media_type="application/x-ndjson") async def restore_generator(): + # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 + host_lock = cmts_config_locks[req.host] + # 1. 嘗試取得全域寫入鎖 (避免多人同時打指令) - if cmts_config_lock.locked(): + if host_lock.locked(): yield json.dumps({"status": "error", "message": "❌ 設備目前正由其他使用者進行配置，請稍後再試。"}) + "\n" return - async with cmts_config_lock: + async with host_lock: try: yield json.dumps({"status": "progress", "message": "🔄 正在建立 SSH 安全連線..."}) + "\n" - conn = await asyncssh.connect(req.host, username=req.username, password=req.password, known_hosts=None) - process = await conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') + + # 🌟 [Priority 2 提前修復] 使用 async with 確保連線與 Process 絕對會被釋放 + async with asyncssh.connect(req.host, username=req.username, password=req.password, known_hosts=None) as 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): - output = "" - while True: - try: - chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=timeout) - if not chunk: break - output += chunk - if "--More--" in chunk or "More" in chunk: - process.stdin.write(" ") - await process.stdin.drain() - except asyncio.TimeoutError: - break - return output + async def read_until_quiet(timeout=1.0): + output = "" + while True: + try: + chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=timeout) + if not chunk: break + output += chunk + if "--More--" in chunk or "More" in chunk: + process.stdin.write(" ") + await process.stdin.drain() + except asyncio.TimeoutError: + break + return output - # 2. 進入設定模式 - process.stdin.write("config\n") - await process.stdin.drain() - await read_until_quiet(timeout=1.5) - yield json.dumps({"status": "progress", "message": "✅ 成功進入 Global Configuration 模式"}) + "\n" - - # 3. 逐行寫入並進行 Fail-safe 檢查 - for cmd in commands: - process.stdin.write(f"{cmd}\n") - await process.stdin.drain() - out = await read_until_quiet(timeout=0.2) - - clean_out = re.sub(r'\x1b\[[0-9;]*[mGK]', '', out).strip() - - # 🚨 Fail-safe 攔截機制與安全撤銷 (Rollback) - if "% Invalid" in clean_out or "% Incomplete" in clean_out or "% Ambiguous" in clean_out: - # 1. 先通知前端正在撤銷 - yield json.dumps({ - "status": "progress", - "message": "⚠️ 偵測到無效指令，正在執行 abort 放棄所有變更..." - }) + "\n" - - # 2. 對設備下達 abort 指令 - process.stdin.write("abort\n") + # 2. 進入設定模式 + process.stdin.write("config\n") await process.stdin.drain() - await read_until_quiet(timeout=1.0) # 等待設備處理 abort 並退出 config 模式 - - # 3. 回報最終錯誤並關閉連線 + await read_until_quiet(timeout=1.5) + yield json.dumps({"status": "progress", "message": "✅ 成功進入 Global Configuration 模式"}) + "\n" + + # 3. 逐行寫入並進行 Fail-safe 檢查 + for cmd in commands: + process.stdin.write(f"{cmd}\n") + await process.stdin.drain() + out = await read_until_quiet(timeout=0.2) + + clean_out = re.sub(r'\x1b\[[0-9;]*[mGK]', '', out).strip() + + # 🚨 Fail-safe 攔截機制與安全撤銷 (Rollback) + if "% Invalid" in clean_out or "% Incomplete" in clean_out or "% Ambiguous" in clean_out: + # 1. 先通知前端正在撤銷 + yield json.dumps({ + "status": "progress", + "message": "⚠️ 偵測到無效指令，正在執行 abort 放棄所有變更..." + }) + "\n" + + # 2. 對設備下達 abort 指令 + process.stdin.write("abort\n") + await process.stdin.drain() + await read_until_quiet(timeout=1.0) # 等待設備處理 abort 並退出 config 模式 + + # 3. 回報最終錯誤並關閉連線 + yield json.dumps({ + "status": "error", + "message": f"❌ 寫入中斷且已安全撤銷！失敗指令: {cmd}", + "output": clean_out + }) + "\n" + # 🌟 移除手動 conn.close()，交由 async with 處理 + return + + yield json.dumps({ + "status": "progress", + "message": f"執行: {cmd}", + "output": clean_out + }) + "\n" + + # 稍微讓出控制權，確保串流順暢 + await asyncio.sleep(0.05) + + # 4. 提交變更 (Commit) + yield json.dumps({"status": "progress", "message": "⏳ 正在寫入 Commit 保存設定..."}) + "\n" + process.stdin.write("commit\n") + await process.stdin.drain() + commit_out = await read_until_quiet(timeout=6.0) + clean_commit = re.sub(r'\x1b\[[0-9;]*[mGK]', '', commit_out).strip() + + # 5. 退出並關閉連線 + process.stdin.write("exit\n") + await process.stdin.drain() + # 🌟 移除手動 conn.close()，交由 async with 處理 + yield json.dumps({ - "status": "error", - "message": f"❌ 寫入中斷且已安全撤銷！失敗指令: {cmd}", - "output": clean_out + "status": "success", + "message": "✅ 所有差異還原指令已成功送出並 Commit！", + "output": clean_commit }) + "\n" - conn.close() - return - - yield json.dumps({ - "status": "progress", - "message": f"執行: {cmd}", - "output": clean_out - }) + "\n" - - # 稍微讓出控制權，確保串流順暢 - await asyncio.sleep(0.05) - - # 4. 提交變更 (Commit) - yield json.dumps({"status": "progress", "message": "⏳ 正在寫入 Commit 保存設定..."}) + "\n" - process.stdin.write("commit\n") - await process.stdin.drain() - commit_out = await read_until_quiet(timeout=6.0) - clean_commit = re.sub(r'\x1b\[[0-9;]*[mGK]', '', commit_out).strip() - - # 5. 退出並關閉連線 - process.stdin.write("exit\n") - await process.stdin.drain() - conn.close() - - yield json.dumps({ - "status": "success", - "message": "✅ 所有差異還原指令已成功送出並 Commit！", - "output": clean_commit - }) + "\n" except Exception as e: logger.error(f"還原執行失敗: {str(e)}") yield json.dumps({"status": "error", "message": f"SSH 連線或執行發生例外錯誤: {str(e)}"}) + "\n" return StreamingResponse(restore_generator(), media_type="application/x-ndjson") - diff --git a/routers/config.py b/routers/config.py index f2e044b..daab193 100644 --- a/routers/config.py +++ b/routers/config.py @@ -9,8 +9,8 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel from typing import List from netmiko import ConnectHandler -# 🌟 移除了舊的 load_settings, save_settings，改用專屬的雙軌機制 -from shared import CMTS_DEVICE, cmts_config_lock, parse_cli_to_tree, deep_split_tree, USE_DB +# 🌟 [Priority 1 修復] 引入 cmts_config_locks +from shared import CMTS_DEVICE, cmts_config_locks, parse_cli_to_tree, deep_split_tree, USE_DB from collections import defaultdict from fastapi.responses import StreamingResponse @@ -89,8 +89,9 @@ async def execute_cmts_config(req: ConfigRequest): commands = [cmd.strip() for cmd in req.script.splitlines() if cmd.strip() and not cmd.strip().startswith('!')] async def config_streamer(): - # 🛡️ 繼承原本的非同步鎖，確保同一時間只有一個寫入任務執行 - async with cmts_config_lock: + # 🌟 [Priority 1 修復] 使用依 IP 隔離的鎖 + host_lock = cmts_config_locks[req.host] + async with host_lock: try: yield json.dumps({"status": "progress", "message": f"🔌 準備連線至設備 {req.host}..."}) + "\n" @@ -235,6 +236,7 @@ async def generate_cli(req: GenerateCliRequest): @router.get("/cmts-ds-rf-port-list") async def get_ds_rf_port_list(host: str, username: str, password: str = ""): """獲取設備上所有的 DS RF Port 清單""" + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) @@ -242,7 +244,6 @@ async def get_ds_rf_port_list(host: str, username: str, password: str = ""): net_connect = ConnectHandler(**device) command = 'show running-config | include "cable ds-rf-port"' raw_output = str(net_connect.send_command(command, read_timeout=60)) - net_connect.disconnect() pattern = r"cable ds-rf-port\s+([0-9:/]+)" ports = re.findall(pattern, raw_output) @@ -255,11 +256,15 @@ async def get_ds_rf_port_list(host: str, username: str, password: str = ""): } except Exception as e: return {"status": "error", "message": str(e)} + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-ds-rf-port-config") async def get_ds_rf_port_config(target: str, host: str, username: str, password: str = ""): """獲取特定 DS RF Port 的配置，並轉換為樹狀 JSON""" + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) @@ -267,10 +272,10 @@ async def get_ds_rf_port_config(target: str, host: str, username: str, password: net_connect = ConnectHandler(**device) command = f"show running-config interface cable ds-rf-port {target} | nomore" raw_output = str(net_connect.send_command(command, read_timeout=60)) - net_connect.disconnect() - base_tree = parse_cli_to_tree(raw_output) - perfect_tree = deep_split_tree(base_tree) + # 🌟 [Priority 1 修復] 將 CPU-Bound 任務移至 ThreadPool + base_tree = await asyncio.to_thread(parse_cli_to_tree, raw_output) + perfect_tree = await asyncio.to_thread(deep_split_tree, base_tree) return { "status": "success", @@ -279,11 +284,15 @@ async def get_ds_rf_port_config(target: str, host: str, username: str, password: } except Exception as e: return {"status": "error", "message": str(e)} + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-full-config") async def get_full_config(host: str, username: str, password: str = "", skip_filter: bool = False, config_type: str = "running"): """獲取整台 CMTS 的完整配置，並轉換為大型樹狀 JSON""" + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) @@ -299,13 +308,12 @@ async def get_full_config(host: str, username: str, password: str = "", skip_fil command = "show running-config | nomore" raw_output = str(net_connect.send_command(command, read_timeout=180)) - net_connect.disconnect() - clean_output = re.sub(r"^(?:Building configuration\.\.\.|Current configuration.*?)\n+", "", raw_output, flags=re.IGNORECASE | re.MULTILINE) clean_output = clean_output.lstrip() - base_tree = parse_cli_to_tree(clean_output) - perfect_tree = deep_split_tree(base_tree) + # 🌟 [Priority 1 修復] 將 CPU-Bound 任務移至 ThreadPool + base_tree = await asyncio.to_thread(parse_cli_to_tree, clean_output) + perfect_tree = await asyncio.to_thread(deep_split_tree, base_tree) # ========================================== # 🌟 深度過濾攔截器：改用雙軌過濾器讀取邏輯 @@ -331,12 +339,16 @@ async def get_full_config(host: str, username: str, password: str = "", skip_fil return {"status": "success", "data": perfect_tree} except Exception as e: return {"status": "error", "message": str(e)} + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-full-config/stream") async def get_full_config_stream(host: str, username: str, password: str = "", skip_filter: bool = False, config_type: str = "running"): """獲取整台 CMTS 的完整配置 (串流進度回報版)""" async def config_streamer(): + net_connect = None try: # 1. 廣播：正在連線 yield json.dumps({"status": "progress", "message": f"🔌 正在建立 SSH 連線至 {host}..."}) + "\n" @@ -360,8 +372,6 @@ async def get_full_config_stream(host: str, username: str, password: str = "", s command = "show running-config | nomore" raw_output = await asyncio.to_thread(net_connect.send_command, command, read_timeout=180) - await asyncio.to_thread(net_connect.disconnect) - # 3. 廣播：正在解析 yield json.dumps({"status": "progress", "message": "🧩 正在解析配置並建構樹狀圖結構..."}) + "\n" await asyncio.sleep(0.1) @@ -369,8 +379,9 @@ async def get_full_config_stream(host: str, username: str, password: str = "", s clean_output = re.sub(r"^(?:Building configuration\.\.\.|Current configuration.*?)\n+", "", raw_output, flags=re.IGNORECASE | re.MULTILINE) clean_output = clean_output.lstrip() - base_tree = parse_cli_to_tree(clean_output) - perfect_tree = deep_split_tree(base_tree) + # 🌟 [Priority 1 修復] 將 CPU-Bound 任務移至 ThreadPool + base_tree = await asyncio.to_thread(parse_cli_to_tree, clean_output) + perfect_tree = await asyncio.to_thread(deep_split_tree, base_tree) # ========================================== # 🌟 深度過濾攔截器 (維持原樣) @@ -399,6 +410,10 @@ async def get_full_config_stream(host: str, username: str, password: str = "", s except Exception as e: yield json.dumps({"status": "error", "message": f"載入失敗: {str(e)}"}) + "\n" + finally: + # 🌟 [Priority 1 修復] 確保背景連線被正確釋放 + if net_connect: + await asyncio.to_thread(net_connect.disconnect) # 回傳 NDJSON 串流格式 return StreamingResponse(config_streamer(), media_type="application/x-ndjson") diff --git a/routers/query.py b/routers/query.py index 23dd8ef..c6a2f80 100644 --- a/routers/query.py +++ b/routers/query.py @@ -30,26 +30,32 @@ def parse_cm_output(raw_text: str) -> list: @router.get("/cable-modems") async def get_cable_modems(): + net_connect = None try: net_connect = ConnectHandler(**CMTS_DEVICE) raw_output = str(net_connect.send_command("show cable modem")) - net_connect.disconnect() structured_data = parse_cm_output(raw_output) return {"status": "success", "total_count": len(structured_data), "data": structured_data} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Connection Error: {str(e)}") + finally: + if net_connect: + net_connect.disconnect() @router.get("/configuration") async def get_configuration(): + net_connect = None try: net_connect = ConnectHandler(**CMTS_DEVICE) net_connect.send_command_timing("config") raw_output = net_connect.send_command("show full-configuration | nomore", expect_string=r"#", read_timeout=120) net_connect.send_command_timing("exit") - net_connect.disconnect() return {"status": "success", "data": raw_output} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Error: {str(e)}") + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-query") async def get_cmts_query( @@ -59,6 +65,7 @@ async def get_cmts_query( username: str = Query(...), password: str = Query(...) ): + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) @@ -107,14 +114,17 @@ async def get_cmts_query( cli_command = commands[query_type] + " | nomore" net_connect = ConnectHandler(**device) raw_output = net_connect.send_command(cli_command, read_timeout=15) - net_connect.disconnect() return {"status": "success", "command": cli_command, "data": raw_output} except Exception as e: raise HTTPException(status_code=500, detail=f"CMTS Query Error: {str(e)}") + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-mac-domain-config") async def get_mac_domain_config(target: str, host: str, username: str, password: str): + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) @@ -122,7 +132,6 @@ async def get_mac_domain_config(target: str, host: str, username: str, password: cli_command = f"show running-config cable mac-domain {target.strip()} | nomore" net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command(cli_command, read_timeout=15)) - net_connect.disconnect() # 💡 正名工程：更新字典的 Key 為一致性的命名 config = { @@ -178,6 +187,9 @@ async def get_mac_domain_config(target: str, host: str, username: str, password: return {"status": "success", "data": config} except Exception as e: raise HTTPException(status_code=500, detail=f"Parse Error: {str(e)}") + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-mac-domain-list") async def get_cmts_mac_domain_list( @@ -185,13 +197,13 @@ async def get_cmts_mac_domain_list( username: str = Query(...), password: str = Query(...) ): + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command_timing("show running-config cable mac-domain ?")) - net_connect.disconnect() import re matches = re.findall(r'\b\d+:\d+/\d+\.\d+\b', raw_output) @@ -200,6 +212,9 @@ async def get_cmts_mac_domain_list( return {"status": "success", "data": mac_domains} except Exception as e: return {"status": "error", "message": str(e)} + finally: + if net_connect: + net_connect.disconnect() @router.get("/cmts-version") async def get_cmts_version( @@ -207,13 +222,13 @@ async def get_cmts_version( username: str = Query(...), password: str = Query(...) ): + net_connect = None try: device = CMTS_DEVICE.copy() device.update({'host': host, 'username': username, 'password': password}) net_connect = ConnectHandler(**device) raw_output = str(net_connect.send_command("show version", read_timeout=15)) - net_connect.disconnect() import re # 🌟 使用 Regex 尋找 infra 或 vcmts-cd-0 後面的版本號 @@ -226,3 +241,6 @@ async def get_cmts_version( except Exception as e: return {"status": "error", "message": f"CMTS Connection Error: {str(e)}"} + finally: + if net_connect: + net_connect.disconnect() diff --git a/shared.py b/shared.py index 8870c82..cf6f59d 100644 --- a/shared.py +++ b/shared.py @@ -258,5 +258,5 @@ 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) -# 全域非同步鎖：確保同一時間只有一人能執行設定寫入 -cmts_config_lock = asyncio.Lock() +# 🌟 [Priority 1 修復] 將全域非同步鎖改為依設備 IP (Host) 隔離的鎖 +cmts_config_locks = defaultdict(asyncio.Lock)