diff --git a/gns3server/agent/__init__.py b/gns3server/agent/__init__.py index dee2aaed1..082c6dd4a 100644 --- a/gns3server/agent/__init__.py +++ b/gns3server/agent/__init__.py @@ -57,7 +57,7 @@ except ImportError as e: AI_COPILOT_AVAILABLE = False # Provide stub functions that raise helpful errors - async def get_project_agent_manager(): + async def get_project_agent_manager() -> "ProjectAgentManager": """ Get the global ProjectAgentManager singleton instance. @@ -68,7 +68,7 @@ except ImportError as e: "AI Copilot is not available. Install AI dependencies with: pip install gns3-server[ai-features]" ) - class ProjectAgentManager: + class ProjectAgentManager: # type: ignore[no-redef] """ Stub class for ProjectAgentManager when AI dependencies are not installed. """ diff --git a/gns3server/agent/gns3_copilot/agent/context_manager.py b/gns3server/agent/gns3_copilot/agent/context_manager.py index 79c526dbf..5337c3d95 100644 --- a/gns3server/agent/gns3_copilot/agent/context_manager.py +++ b/gns3server/agent/gns3_copilot/agent/context_manager.py @@ -140,7 +140,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int: for tool in tools: try: # Build tool schema in OpenAI format - tool_schema = { + tool_schema: dict[str, Any] = { "type": "function", "function": { "name": tool.name, diff --git a/gns3server/agent/gns3_copilot/agent/model_factory.py b/gns3server/agent/gns3_copilot/agent/model_factory.py index cfdc26854..f790646bd 100644 --- a/gns3server/agent/gns3_copilot/agent/model_factory.py +++ b/gns3server/agent/gns3_copilot/agent/model_factory.py @@ -110,7 +110,7 @@ def create_base_model( try: # Prepare parameters for init_chat_model - init_params = { + init_params: dict[str, Any] = { "model": config_vars["model_name"], "model_provider": config_vars["model_provider"], "api_key": config_vars["api_key"], @@ -177,7 +177,7 @@ def create_title_model( try: # Prepare parameters for init_chat_model - init_params = { + init_params: dict[str, Any] = { "model": config_vars["model_name"], "model_provider": config_vars["model_provider"], "api_key": config_vars["api_key"], diff --git a/gns3server/agent/gns3_copilot/chat_sessions_repository.py b/gns3server/agent/gns3_copilot/chat_sessions_repository.py index 4d18d8852..41006f299 100644 --- a/gns3server/agent/gns3_copilot/chat_sessions_repository.py +++ b/gns3server/agent/gns3_copilot/chat_sessions_repository.py @@ -125,7 +125,7 @@ class ChatSessionsRepository: project_id: str, title: str = "New Conversation", copilot_mode: Optional[str] = None, - ) -> ChatSession: + ) -> Optional[ChatSession]: """ Create a new chat session. @@ -158,6 +158,8 @@ class ChatSessionsRepository: session_id = cursor.lastrowid log.info("Created chat session: id=%s, thread_id=%s, copilot_mode=%s", session_id, thread_id, copilot_mode) + if session_id is None: + return None return await self.get_session_by_id(session_id) async def get_session_by_id(self, session_id: int) -> Optional[ChatSession]: @@ -214,7 +216,7 @@ class ChatSessionsRepository: List of ChatSession """ query = "SELECT * FROM chat_sessions" - params = [] + params: List[Any] = [] conditions = [] if user_id: @@ -268,7 +270,7 @@ class ChatSessionsRepository: Updated ChatSession or None """ updates = [] - params = [] + params: List[Any] = [] now = datetime.utcnow().isoformat() diff --git a/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py b/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py index b0934c201..7f6ca8c95 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py +++ b/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py @@ -162,14 +162,14 @@ def build_gns3_ctx(jwt_token: str | None = None, url: str | None = None) -> dict } -def _filter_node_response(node: dict, fields: list[str] = None) -> dict: +def _filter_node_response(node: dict, fields: list[str] | None = None) -> dict: """Filter node response to only include requested fields.""" if not fields: fields = ["node_id", "name", "node_type", "status", "console"] return {k: node[k] for k in fields if k in node} -def _filter_link_response(link: dict, fields: list[str] = None) -> dict: +def _filter_link_response(link: dict, fields: list[str] | None = None) -> dict: """Filter link response to only include requested fields.""" if not fields: fields = LINK_DEFAULT_FIELDS @@ -278,7 +278,7 @@ def _batch_lifecycle(project_id, node_ids, action, conn, action_label): return list(pool.map(_act, node_ids)) -def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} @@ -296,7 +296,7 @@ def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict return {"message": f"Node {node_id} started", "node_id": node_id} -def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} @@ -314,7 +314,7 @@ def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[ return {"message": f"Node {node_id} stopped", "node_id": node_id} -def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} @@ -332,7 +332,7 @@ def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di return {"message": f"Node {node_id} suspended", "node_id": node_id} -def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: @@ -401,7 +401,7 @@ def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic return _filter_node_response(resp, fields) -def delete_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def delete_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} @@ -706,7 +706,7 @@ def available_filters_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) return conn.http_call("get", url).json() -def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} @@ -762,7 +762,7 @@ def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic return _filter_link_response(resp, fields) -def delete_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def delete_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} @@ -809,7 +809,7 @@ def update_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic # ── Link capture / reset handlers ────────────────────────────────────── -def reset_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: +def reset_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]: project_id = params.get("project_id") if not project_id: return {"error": "project_id is required"} diff --git a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py index c57dc8fee..32a4e2e45 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py +++ b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py @@ -347,7 +347,7 @@ def get_gns3_server_host() -> str: host = get_gns3_server_host() print(f"GNS3 server host: {host}") """ - url = _detect_url_for_api() + url = _detect_url_for_api() or DEFAULT_GNS3_URL # Extract host from URL # URL format: protocol://host:port diff --git a/gns3server/agent/gns3_copilot/gns3_client/project_inventory.py b/gns3server/agent/gns3_copilot/gns3_client/project_inventory.py index b1f958767..aeefd304a 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/project_inventory.py +++ b/gns3server/agent/gns3_copilot/gns3_client/project_inventory.py @@ -49,7 +49,7 @@ def build_nodes_inventory(nodes: list[dict[str, Any]], server_host: str | None) """ inventory: dict[str, Any] = {} for n in nodes: - inventory[n.get("name")] = { + inventory[n.get("name") or ""] = { "server": server_host, "name": n.get("name"), "node_id": n.get("node_id"), @@ -68,7 +68,7 @@ def build_nodes_inventory(nodes: list[dict[str, Any]], server_host: str | None) return inventory -def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]) -> list[dict[str, str]]: +def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]) -> list[dict[str, Any]]: """ Build a human/LLM-friendly link list resolving node and port names. @@ -76,7 +76,7 @@ def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]] Links whose endpoints cannot be resolved are skipped, mirroring the original Project.links_summary behavior. """ - summary: list[dict[str, str]] = [] + summary: list[dict[str, Any]] = [] for link in links: if not link.get("nodes"): continue diff --git a/gns3server/agent/gns3_copilot/skills/loader.py b/gns3server/agent/gns3_copilot/skills/loader.py index 655d76263..0295e2520 100644 --- a/gns3server/agent/gns3_copilot/skills/loader.py +++ b/gns3server/agent/gns3_copilot/skills/loader.py @@ -248,7 +248,7 @@ class SkillsLoader: continue # Use device_type from YAML content as the key # Fallback to filename stem if device_type not present - skill_key = skill_data.get("device_type") if isinstance(skill_data, dict) else None + skill_key = skill_data.get("device_type") if not skill_key: skill_key = yaml_file.stem logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key") diff --git a/gns3server/agent/gns3_copilot/skills/manager.py b/gns3server/agent/gns3_copilot/skills/manager.py index 473fb2b09..1a511ec29 100644 --- a/gns3server/agent/gns3_copilot/skills/manager.py +++ b/gns3server/agent/gns3_copilot/skills/manager.py @@ -37,8 +37,10 @@ from typing import Optional, Dict, Any try: import git + + GIT_AVAILABLE = True except ImportError: - git = None + GIT_AVAILABLE = False from gns3server.config import Config from .loader import SkillsLoader @@ -66,7 +68,7 @@ class SkillsManager: - Version tracking """ - def __init__(self, repo_url: str = None, branch: str = "main", auto_update: bool = False): + def __init__(self, repo_url: Optional[str] = None, branch: str = "main", auto_update: bool = False): """ Initialize the skills manager. @@ -90,7 +92,7 @@ class SkillsManager: self._repo: Optional["git.Repo"] = None self._prompt_count = 0 - if git is None: + if not GIT_AVAILABLE: logger.warning("GitPython is not installed. Skills management features will be limited.") def initialize(self) -> bool: @@ -134,7 +136,7 @@ class SkillsManager: - Up to date → nothing - Network error → use existing files, log warning """ - if git is None or self._repo is None: + if not GIT_AVAILABLE or self._repo is None: return # Check for uncommitted changes @@ -173,7 +175,7 @@ class SkillsManager: Returns: True if successful, False otherwise """ - if git is None: + if not GIT_AVAILABLE: logger.error("GitPython is not installed. Cannot clone repository.") return False @@ -347,7 +349,7 @@ class SkillsManager: Returns: Commit hash as string, or empty string if not available """ - if git is None or self._repo is None: + if not GIT_AVAILABLE or self._repo is None: try: self._repo = git.Repo(self.local_path) except Exception: @@ -409,7 +411,7 @@ class SkillsManager: Returns: True if successful, False otherwise """ - if git is None: + if not GIT_AVAILABLE: logger.error("GitPython is not installed. Cannot rollback.") return False @@ -436,7 +438,7 @@ class SkillsManager: Returns: List of commit information dictionaries """ - if git is None: + if not GIT_AVAILABLE: return [] try: diff --git a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py index db215c6bf..ed15f0589 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -78,7 +78,7 @@ try: # so we use importlib to get the actual module import importlib - sd = importlib.import_module("netmiko.ssh_dispatcher") + sd: Any = importlib.import_module("netmiko.ssh_dispatcher") # Recalculate platforms lists to include custom device types sd.platforms = list(sd.CLASS_MAPPER.keys()) @@ -196,7 +196,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Validate input device_configs_list, project_id = self._validate_tool_input(tool_input) - if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]: + if len(device_configs_list) > 0 and "error" in device_configs_list[0]: return device_configs_list # Filter forbidden commands and store blocked commands info diff --git a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py index 8c4a5281d..a8eb41bc4 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -78,7 +78,7 @@ try: # so we use importlib to get the actual module import importlib - sd = importlib.import_module("netmiko.ssh_dispatcher") + sd: Any = importlib.import_module("netmiko.ssh_dispatcher") # Recalculate platforms lists to include custom device types sd.platforms = list(sd.CLASS_MAPPER.keys()) @@ -199,7 +199,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): # Validate input device_configs_list, project_id = self._validate_tool_input(tool_input) - if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]: + if len(device_configs_list) > 0 and "error" in device_configs_list[0]: return device_configs_list # Filter forbidden commands and store blocked commands info diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py index 8f926586b..ff11c6a44 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py @@ -222,6 +222,8 @@ class GNS3LinkTool(BaseTool): }, gns3_ctx, ) + if isinstance(link_resp, list): + raise RuntimeError("Unexpected batch response when creating a single link") if "error" in link_resp: raise RuntimeError(link_resp["error"]) diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py index a511de21f..bdbc0bf42 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py @@ -230,6 +230,8 @@ class GNS3CreateNodeTool(BaseTool): }, gns3_ctx, ) + if isinstance(created, list): + raise RuntimeError("Unexpected batch response when creating a single node") if "error" in created: raise RuntimeError(created["error"]) node_info = { diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py index d94a0fc1d..be7af824f 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py @@ -286,7 +286,7 @@ class GNS3PacketFilterTool(BaseTool): # Check if output contains "Invalid" indicating syntax error if "Invalid" in result.stdout or "Invalid" in result.stderr: - error_lines = [] + error_lines: list[str] = [] if "Invalid" in result.stderr: error_lines.extend(line for line in result.stderr.split("\n") if "Invalid" in line) if "Invalid" in result.stdout: diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py index 8e98c7480..4086cde1b 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py @@ -136,7 +136,11 @@ class GNS3StartNodeTool(BaseTool): results = [] known_ids = [nid for nid in node_ids if nid in nodes_by_id] start_results = start_node_handler({"project_id": project_id, "node_ids": known_ids}, gns3_ctx) - start_errors = {r["node_id"]: r.get("error") for r in start_results if r.get("status") == "error"} + start_errors = ( + {r["node_id"]: r.get("error") for r in start_results if r.get("status") == "error"} + if isinstance(start_results, list) + else {} + ) # Get immediate status (likely 'starting' or 'stopped') — one call listing = get_nodes_handler({"project_id": project_id}, gns3_ctx) diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py index 3c0d8ef2c..d6965f19b 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py @@ -124,7 +124,11 @@ class GNS3StopNodeTool(BaseTool): results = [] known_ids = [nid for nid in node_ids if nid in nodes_by_id] stop_results = stop_node_handler({"project_id": project_id, "node_ids": known_ids}, gns3_ctx) - stop_errors = {r["node_id"]: r.get("error") for r in stop_results if r.get("status") == "error"} + stop_errors = ( + {r["node_id"]: r.get("error") for r in stop_results if r.get("status") == "error"} + if isinstance(stop_results, list) + else {} + ) # Get updated status — one call listing = get_nodes_handler({"project_id": project_id}, gns3_ctx) diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py index 74f9ba883..47a0cc4b1 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py @@ -130,7 +130,11 @@ class GNS3SuspendNodeTool(BaseTool): results = [] known_ids = [nid for nid in node_ids if nid in nodes_by_id] suspend_results = suspend_node_handler({"project_id": project_id, "node_ids": known_ids}, gns3_ctx) - suspend_errors = {r["node_id"]: r.get("error") for r in suspend_results if r.get("status") == "error"} + suspend_errors = ( + {r["node_id"]: r.get("error") for r in suspend_results if r.get("status") == "error"} + if isinstance(suspend_results, list) + else {} + ) # Get updated status — one call listing = get_nodes_handler({"project_id": project_id}, gns3_ctx) diff --git a/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py b/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py index b9c35d141..c5f30bf79 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py +++ b/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py @@ -143,7 +143,8 @@ class PacketAnalysisTool(BaseTool): stderr=subprocess.DEVNULL, text=True, ) - tshark.stdout.close() + if tshark.stdout is not None: + tshark.stdout.close() stdout, _ = grep.communicate(timeout=30) tshark.wait(timeout=5) except Exception as e: diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py index 3a8789cda..1315730c7 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py +++ b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py @@ -68,7 +68,7 @@ try: # CRITICAL: Update netmiko.ssh_dispatcher platforms lists import importlib - sd = importlib.import_module("netmiko.ssh_dispatcher") + sd: Any = importlib.import_module("netmiko.ssh_dispatcher") # Recalculate platforms lists to include custom device types sd.platforms = list(sd.CLASS_MAPPER.keys()) @@ -179,7 +179,7 @@ class VPCSCommands(BaseTool): # Validate input device_configs_list, project_id = self._validate_tool_input(tool_input) - if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]: + if len(device_configs_list) > 0 and "error" in device_configs_list[0]: return device_configs_list # Create a mapping of device names to their commands @@ -386,7 +386,7 @@ class VPCSCommands(BaseTool): def _prepare_device_hosts_data( self, device_configs_list: list[dict[str, Any]], - project_id: str, + project_id: str | None, jwt_token: str | None = None, url: str | None = None, ) -> dict[str, dict[str, Any]]: @@ -415,7 +415,7 @@ class VPCSCommands(BaseTool): device_ports = get_device_ports_from_topology(device_names, project_id=project_id, jwt_token=jwt_token, url=url) # Build Nornir inventory hosts data - hosts_data = {} + hosts_data: dict[str, Any] = {} for device_name in device_names: if device_name not in device_ports: logger.error("Device '%s' not found in topology", device_name) diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py index 3e951f10a..d4f7b1a38 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py @@ -169,7 +169,7 @@ def group_devices_by_brand(device_types: List[str]) -> Dict[str, List[str]]: Returns: Dictionary mapping brand names to device type lists """ - brands = {} + brands: dict[str, list[str]] = {} for device_type in device_types: brand = extract_brand_name(device_type) diff --git a/gns3server/agent/gns3_copilot/utils/message_converters.py b/gns3server/agent/gns3_copilot/utils/message_converters.py index 8bf18f668..367b2d956 100644 --- a/gns3server/agent/gns3_copilot/utils/message_converters.py +++ b/gns3server/agent/gns3_copilot/utils/message_converters.py @@ -37,6 +37,7 @@ from typing import Dict from langchain_core.messages import AIMessage from langchain_core.messages import HumanMessage from langchain_core.messages import SystemMessage +from langchain_core.messages import ToolCall from langchain_core.messages import ToolMessage @@ -86,7 +87,7 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]: tool_calls = [] for tc in lc_message.tool_calls: # Convert to dict if it's an object - tc_dict = tc if isinstance(tc, dict) else tc.model_dump() + tc_dict = tc tool_calls.append( { "id": tc_dict.get("id", str(uuid.uuid4())), @@ -139,7 +140,7 @@ def convert_openai_to_langchain(msg: Dict[str, Any]): # Restore tool calls if present if "tool_calls" in msg and msg["tool_calls"]: - tool_calls = [] + tool_calls: list[ToolCall] = [] for tc in msg["tool_calls"]: tool_calls.append( { diff --git a/gns3server/agent/gns3_copilot/utils/parse_tool_content.py b/gns3server/agent/gns3_copilot/utils/parse_tool_content.py index 4ea17912d..350a6bf50 100644 --- a/gns3server/agent/gns3_copilot/utils/parse_tool_content.py +++ b/gns3server/agent/gns3_copilot/utils/parse_tool_content.py @@ -214,9 +214,7 @@ def parse_tool_content( return content # Handle unsupported types - error_msg = ( # type: ignore[unreachable] - f"Content must be str, dict, list, int, float, bool, or None, got {type(content).__name__}" - ) + error_msg = f"Content must be str, dict, list, int, float, bool, or None, got {type(content).__name__}" logger.error(error_msg) if strict_mode: diff --git a/gns3server/agent/mcp/appliances.py b/gns3server/agent/mcp/appliances.py index 45a9b472a..6e3a3ecf5 100644 --- a/gns3server/agent/mcp/appliances.py +++ b/gns3server/agent/mcp/appliances.py @@ -108,7 +108,7 @@ def install_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) if version: request_params["version"] = version response = conn.http_call("post", url, params=request_params) - result = {"message": f"Appliance {appliance_id} installed"} + result: dict[str, Any] = {"message": f"Appliance {appliance_id} installed"} if response.content: # the install endpoint returns the created template (201); tolerate an # empty body in case an older server still replies with 204 diff --git a/gns3server/agent/web_wireshark/docker_client.py b/gns3server/agent/web_wireshark/docker_client.py index d27873fa5..00f62edd2 100644 --- a/gns3server/agent/web_wireshark/docker_client.py +++ b/gns3server/agent/web_wireshark/docker_client.py @@ -23,9 +23,16 @@ This module handles Docker API operations via aiohttp Unix socket. import asyncio import logging +import sys +from typing import Any import aiohttp +if sys.version_info >= (3, 11): + from asyncio import timeout as async_timeout +else: + from async_timeout import timeout as async_timeout + from gns3server.utils import parse_version logger = logging.getLogger(__name__) @@ -132,7 +139,7 @@ class DockerHTTPClient: url = f"http://docker/v{self._api_version}/{endpoint}" try: - async with asyncio.timeout(self.REQUEST_TIMEOUT): + async with async_timeout(self.REQUEST_TIMEOUT): async with session.request(method, url, **kwargs) as response: if response.status >= 300: error_text = await response.text() @@ -157,9 +164,9 @@ class DockerHTTPClient: return await self._request(method, endpoint, check_connection=False, **kwargs) raise - async def create_network(self, name: str, driver: str = "bridge", subnet: str = None): + async def create_network(self, name: str, driver: str = "bridge", subnet: str | None = None): """Create Docker network.""" - data = {"Name": name, "Driver": driver} + data: dict[str, Any] = {"Name": name, "Driver": driver} if subnet: data["IPAM"] = {"Config": [{"Subnet": subnet}]} await self._request("POST", "networks/create", json=data) @@ -175,7 +182,7 @@ class DockerHTTPClient: async def create_container(self, name: str, image: str, **kwargs): """Create container.""" - data = {"Image": image, "name": name, "HostConfig": {}, "NetworkingConfig": {}} + data: dict[str, Any] = {"Image": image, "name": name, "HostConfig": {}, "NetworkingConfig": {}} # Handle network config if "network" in kwargs: @@ -248,7 +255,7 @@ class DockerHTTPClient: url = f"http://docker/v{self._api_version}/containers/{container_name}/top?ps_args=aux" try: - async with asyncio.timeout(self.REQUEST_TIMEOUT): + async with async_timeout(self.REQUEST_TIMEOUT): async with session.get(url) as response: if response.status >= 300: error_text = await response.text() diff --git a/gns3server/agent/web_wireshark/manage_wireshark.py b/gns3server/agent/web_wireshark/manage_wireshark.py index 559eaadd9..5203a98d9 100755 --- a/gns3server/agent/web_wireshark/manage_wireshark.py +++ b/gns3server/agent/web_wireshark/manage_wireshark.py @@ -81,10 +81,7 @@ from gns3server.utils.uuid_validator import validate_uuid if __name__ == "__main__": sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) -try: - from gns3server.agent.web_wireshark.manager import WebWiresharkManager -except ImportError: - from manager import WebWiresharkManager +from gns3server.agent.web_wireshark.manager import WebWiresharkManager logger = logging.getLogger(__name__) diff --git a/gns3server/agent/web_wireshark/manager.py b/gns3server/agent/web_wireshark/manager.py index e17cf65c0..16f057d75 100644 --- a/gns3server/agent/web_wireshark/manager.py +++ b/gns3server/agent/web_wireshark/manager.py @@ -84,7 +84,7 @@ class WebWiresharkManager: logger.warning(f"Container {container_id[:12]} health check failed: {e}") return False - async def _exec_in_container(self, container_id: str, command: str, timeout: int = None) -> tuple: + async def _exec_in_container(self, container_id: str, command: str, timeout: Optional[int] = None) -> tuple: """Execute command in container with timeout. Args: @@ -197,7 +197,7 @@ class WebWiresharkManager: stdout, _ = await proc.communicate() # Build parent->children mapping - children_map = {} + children_map: dict[int, list[int]] = {} for line in stdout.decode().strip().split("\n"): parts = line.split(None, 2) if len(parts) < 3: @@ -326,7 +326,7 @@ class WebWiresharkManager: return # Build parent->children mapping and collect process info - children_map = {} # ppid -> [pid] + children_map: dict[int, list[int]] = {} # ppid -> [pid] process_info = {} # pid -> (ppid, command) for line in stdout.decode().strip().split("\n"): @@ -480,7 +480,7 @@ class WebWiresharkManager: logger.debug(f"Cannot get URL from Config: {e}") return None - async def _get_container_gateway_ip(self, container_id: str = None) -> Optional[str]: + async def _get_container_gateway_ip(self, container_id: Optional[str] = None) -> Optional[str]: """Get the Docker bridge gateway IP for container to access host. Args: @@ -519,7 +519,7 @@ class WebWiresharkManager: return None - async def get_container_ip(self, container_name: str, container_id: str = None) -> Optional[str]: + async def get_container_ip(self, container_name: str, container_id: Optional[str] = None) -> Optional[str]: """Get the container IP address in the wireshark network. Args: @@ -626,7 +626,7 @@ class WebWiresharkManager: project_id: str, image: str = "gns3/web-wireshark:latest", memory: str = "2g", - memory_swap: str = None, + memory_swap: Optional[str] = None, cpus: float = 1.0, pids_limit: int = 1000, ) -> str: @@ -935,10 +935,10 @@ class WebWiresharkManager: project_id: str, link_id: str, jwt_token: str, - capture_stream_url: str = None, + capture_stream_url: Optional[str] = None, image: str = "gns3/web-wireshark:latest", memory: str = "2g", - memory_swap: str = None, + memory_swap: Optional[str] = None, cpus: float = 1.0, pids_limit: int = 1000, ): diff --git a/gns3server/agent/web_wireshark/stats.py b/gns3server/agent/web_wireshark/stats.py index 78dc7b0c0..de9b66091 100644 --- a/gns3server/agent/web_wireshark/stats.py +++ b/gns3server/agent/web_wireshark/stats.py @@ -7,7 +7,7 @@ about Web Wireshark containers and sessions. import logging import subprocess -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) @@ -30,7 +30,7 @@ async def collect_webwireshark_stats(projects: List) -> Dict: """ from .manager import WebWiresharkManager - stats = {"total_containers": 0, "running_containers": 0, "active_sessions": 0, "containers": []} + stats: Dict[str, Any] = {"total_containers": 0, "running_containers": 0, "active_sessions": 0, "containers": []} # Create a single manager instance and reuse it manager = WebWiresharkManager() diff --git a/gns3server/api/routes/compute/atm_switch_nodes.py b/gns3server/api/routes/compute/atm_switch_nodes.py index 65bc1157d..1282a162d 100644 --- a/gns3server/api/routes/compute/atm_switch_nodes.py +++ b/gns3server/api/routes/compute/atm_switch_nodes.py @@ -18,6 +18,7 @@ API routes for ATM switch nodes. """ +from typing import Any import os from fastapi import APIRouter, Depends, Body, Path, status, HTTPException @@ -29,7 +30,9 @@ from gns3server import schemas from gns3server.compute.dynamips import Dynamips from gns3server.compute.dynamips.nodes.atm_switch import ATMSwitch -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or ATM switch node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or ATM switch node"} +} router = APIRouter(responses=responses) @@ -57,13 +60,13 @@ async def create_atm_switch(project_id: UUID, node_data: schemas.ATMSwitchCreate # Use the Dynamips ATM switch to simulate this node dynamips_manager = Dynamips.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await dynamips_manager.create_node( - node_data.get("name"), + data.get("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="atm_switch", - mappings=node_data.get("mappings"), + mappings=data.get("mappings"), ) return node.asdict() @@ -97,11 +100,11 @@ async def update_atm_switch( Update an ATM switch node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - if "name" in node_data and node.name != node_data["name"]: - await node.set_name(node_data["name"]) - if "mappings" in node_data: - node.mappings = node_data["mappings"] + data = jsonable_encoder(node_data, exclude_unset=True) + if "name" in data and node.name != data["name"]: + await node.set_name(data["name"]) + if "mappings" in data: + node.mappings = data["mappings"] node.updated() return node.asdict() diff --git a/gns3server/api/routes/compute/capabilities.py b/gns3server/api/routes/compute/capabilities.py index f472f6231..42154d4e1 100644 --- a/gns3server/api/routes/compute/capabilities.py +++ b/gns3server/api/routes/compute/capabilities.py @@ -39,7 +39,8 @@ def get_capabilities(request: Request) -> dict: node_types.extend(module.node_types()) # record the controller hostname or IP address - request.app.state.controller_host = request.client.host + if request.client: + request.app.state.controller_host = request.client.host return { "version": __version__, diff --git a/gns3server/api/routes/compute/cloud_nodes.py b/gns3server/api/routes/compute/cloud_nodes.py index 7baf8ca17..d0a4ead9e 100644 --- a/gns3server/api/routes/compute/cloud_nodes.py +++ b/gns3server/api/routes/compute/cloud_nodes.py @@ -23,14 +23,16 @@ import os from fastapi import APIRouter, Depends, Path, status, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse -from typing import Union +from typing import Any, Union from uuid import UUID from gns3server import schemas from gns3server.compute.builtin import Builtin from gns3server.compute.builtin.nodes.cloud import Cloud -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or cloud node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or cloud node"} +} router = APIRouter(responses=responses) @@ -57,21 +59,21 @@ async def create_cloud(project_id: UUID, node_data: schemas.CloudCreate) -> sche """ builtin_manager = Builtin.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await builtin_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="cloud", - ports=node_data.get("ports_mapping"), + ports=data.get("ports_mapping"), ) # add the remote console settings - node.remote_console_host = node_data.get("remote_console_host", node.remote_console_host) - node.remote_console_port = node_data.get("remote_console_port", node.remote_console_port) - node.remote_console_type = node_data.get("remote_console_type", node.remote_console_type) - node.remote_console_http_path = node_data.get("remote_console_http_path", node.remote_console_http_path) - node.usage = node_data.get("usage", "") + node.remote_console_host = data.get("remote_console_host", node.remote_console_host) + node.remote_console_port = data.get("remote_console_port", node.remote_console_port) + node.remote_console_type = data.get("remote_console_type", node.remote_console_type) + node.remote_console_http_path = data.get("remote_console_http_path", node.remote_console_http_path) + node.usage = data.get("usage", "") return node.asdict() @@ -90,8 +92,8 @@ async def update_cloud(node_data: schemas.CloudUpdate, node: Cloud = Depends(dep Update a cloud node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - for name, value in node_data.items(): + data = jsonable_encoder(node_data, exclude_unset=True) + for name, value in data.items(): if hasattr(node, name) and getattr(node, name) != value: setattr(node, name, value) node.updated() @@ -178,7 +180,7 @@ async def update_cloud_nio( nio = node.get_nio(port_number) nio.filters.clear() - if nio_data.filters: + if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters: nio.filters = nio_data.filters # NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers. nio.markers = getattr(nio_data, "markers", None) or {} diff --git a/gns3server/api/routes/compute/compute.py b/gns3server/api/routes/compute/compute.py index 1a862787c..c98b1a256 100644 --- a/gns3server/api/routes/compute/compute.py +++ b/gns3server/api/routes/compute/compute.py @@ -163,7 +163,7 @@ def compute_statistics() -> dict: @router.get("/qemu/capabilities") async def get_qemu_capabilities() -> dict: - capabilities = {"kvm": []} + capabilities: dict[str, list] = {"kvm": []} kvms = await Qemu.get_kvm_archs() if kvms: capabilities["kvm"] = kvms diff --git a/gns3server/api/routes/compute/dependencies/authentication.py b/gns3server/api/routes/compute/dependencies/authentication.py index 14ba07f91..0412a4136 100644 --- a/gns3server/api/routes/compute/dependencies/authentication.py +++ b/gns3server/api/routes/compute/dependencies/authentication.py @@ -91,9 +91,9 @@ async def ws_compute_authentication(websocket: WebSocket) -> Union[None, WebSock if not separator: raise invalid_user_credentials_exc - username = secrets.compare_digest(username, server_settings.compute_username) - password = secrets.compare_digest(password, server_settings.compute_password.get_secret_value()) - if not (username and password): + username_valid = secrets.compare_digest(username, server_settings.compute_username) + password_valid = secrets.compare_digest(password, server_settings.compute_password.get_secret_value()) + if not (username_valid and password_valid): raise invalid_user_credentials_exc except HTTPException as e: @@ -101,5 +101,6 @@ async def ws_compute_authentication(websocket: WebSocket) -> Union[None, WebSock websocket_error = {"action": "log.error", "event": {"message": err_msg}} await websocket.send_json(websocket_error) log.error(err_msg) - return await websocket.close(code=1008) + await websocket.close(code=1008) + return None return websocket diff --git a/gns3server/api/routes/compute/docker_nodes.py b/gns3server/api/routes/compute/docker_nodes.py index 6091cf37b..9dee90dda 100644 --- a/gns3server/api/routes/compute/docker_nodes.py +++ b/gns3server/api/routes/compute/docker_nodes.py @@ -24,14 +24,16 @@ from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse from uuid import UUID -from typing import Union +from typing import Any, Union from gns3server import schemas from gns3server.compute.docker import Docker from gns3server.compute.docker.docker_vm import DockerVM from .dependencies.authentication import compute_authentication, ws_compute_authentication -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Docker node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or Docker node"} +} router = APIRouter(responses=responses) @@ -53,13 +55,13 @@ def dep_node(project_id: UUID, node_id: UUID) -> DockerVM: responses={409: {"model": schemas.ErrorMessage, "description": "Could not create Docker node"}}, dependencies=[Depends(compute_authentication)], ) -async def create_docker_node(project_id: UUID, node_data: schemas.DockerCreate) -> schemas.Docker: +async def create_docker_node(project_id: UUID, node_create: schemas.DockerCreate) -> schemas.Docker: """ Create a new Docker node. """ docker_manager = Docker.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + node_data = jsonable_encoder(node_create, exclude_unset=True) container = await docker_manager.create_node( node_data.pop("name"), str(project_id), @@ -124,7 +126,7 @@ def get_docker_node(node: DockerVM = Depends(dep_node)) -> schemas.Docker: @router.put("/{node_id}", response_model=schemas.Docker, dependencies=[Depends(compute_authentication)]) -async def update_docker_node(node_data: schemas.DockerUpdate, node: DockerVM = Depends(dep_node)) -> schemas.Docker: +async def update_docker_node(node_update: schemas.DockerUpdate, node: DockerVM = Depends(dep_node)) -> schemas.Docker: """ Update a Docker node. """ @@ -152,7 +154,7 @@ async def update_docker_node(node_data: schemas.DockerUpdate, node: DockerVM = D ] changed = False - node_data = jsonable_encoder(node_data, exclude_unset=True) + node_data = jsonable_encoder(node_update, exclude_unset=True) for prop in props: # hasattr: startup_config_content only exists on IOLDockerVM if prop in node_data and hasattr(node, prop) and node_data[prop] != getattr(node, prop): diff --git a/gns3server/api/routes/compute/dynamips_nodes.py b/gns3server/api/routes/compute/dynamips_nodes.py index ce1c1b42b..ccfcd58fb 100644 --- a/gns3server/api/routes/compute/dynamips_nodes.py +++ b/gns3server/api/routes/compute/dynamips_nodes.py @@ -23,7 +23,7 @@ import os from fastapi import APIRouter, WebSocket, Body, Depends, status, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse -from typing import List, Union +from typing import Any, List, Optional, Union from uuid import UUID from gns3server.compute.dynamips import Dynamips @@ -32,7 +32,9 @@ from gns3server import schemas from .dependencies.authentication import compute_authentication, ws_compute_authentication -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Dynamips node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or Dynamips node"} +} router = APIRouter(responses=responses) @@ -64,25 +66,26 @@ async def create_router(project_id: UUID, node_data: schemas.DynamipsCreate) -> dynamips_manager = Dynamips.instance() platform = node_data.platform + chassis: Optional[str] if not node_data.chassis and platform in DEFAULT_CHASSIS: chassis = DEFAULT_CHASSIS[platform] else: chassis = node_data.chassis - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) vm = await dynamips_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), - dynamips_id=node_data.get("dynamips_id"), + data.get("node_id"), + dynamips_id=data.get("dynamips_id"), platform=platform, - console=node_data.get("console"), - console_type=node_data.get("console_type", "telnet"), - aux=node_data.get("aux"), - aux_type=node_data.pop("aux_type", "none"), + console=data.get("console"), + console_type=data.get("console_type", "telnet"), + aux=data.get("aux"), + aux_type=data.pop("aux_type", "none"), chassis=chassis, node_type="dynamips", ) - await dynamips_manager.update_vm_settings(vm, node_data) + await dynamips_manager.update_vm_settings(vm, data) return vm.asdict() diff --git a/gns3server/api/routes/compute/ethernet_hub_nodes.py b/gns3server/api/routes/compute/ethernet_hub_nodes.py index 350105789..150d18e64 100644 --- a/gns3server/api/routes/compute/ethernet_hub_nodes.py +++ b/gns3server/api/routes/compute/ethernet_hub_nodes.py @@ -18,6 +18,7 @@ API routes for Ethernet hub nodes. """ +from typing import Any import os from fastapi import APIRouter, Depends, Body, Path, status, HTTPException @@ -29,7 +30,9 @@ from gns3server.compute.dynamips import Dynamips from gns3server.compute.dynamips.nodes.ethernet_hub import EthernetHub from gns3server import schemas -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet hub node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet hub node"} +} router = APIRouter(responses=responses) @@ -57,13 +60,13 @@ async def create_ethernet_hub(project_id: UUID, node_data: schemas.EthernetHubCr # Use the Dynamips Ethernet hub to simulate this node dynamips_manager = Dynamips.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await dynamips_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="ethernet_hub", - ports=node_data.get("ports_mapping"), + ports=data.get("ports_mapping"), ) return node.asdict() @@ -97,11 +100,11 @@ async def update_ethernet_hub( Update an Ethernet hub. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - if "name" in node_data and node.name != node_data["name"]: - await node.set_name(node_data["name"]) - if "ports_mapping" in node_data: - node.ports_mapping = node_data["ports_mapping"] + data = jsonable_encoder(node_data, exclude_unset=True) + if "name" in data and node.name != data["name"]: + await node.set_name(data["name"]) + if "ports_mapping" in data: + node.ports_mapping = data["ports_mapping"] node.updated() return node.asdict() diff --git a/gns3server/api/routes/compute/ethernet_switch_nodes.py b/gns3server/api/routes/compute/ethernet_switch_nodes.py index bedfac0b6..9c7736e9c 100644 --- a/gns3server/api/routes/compute/ethernet_switch_nodes.py +++ b/gns3server/api/routes/compute/ethernet_switch_nodes.py @@ -22,6 +22,7 @@ through uBridge's ``brctl`` module (see ``gns3server.compute.builtin.nodes.ethernet_switch``). """ +from typing import Any import os from fastapi import APIRouter, Depends, Body, Path, status, HTTPException @@ -33,7 +34,9 @@ from gns3server.compute.builtin import Builtin from gns3server.compute.builtin.nodes.ethernet_switch import EthernetSwitch from gns3server import schemas -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet switch node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet switch node"} +} router = APIRouter(responses=responses) @@ -60,17 +63,17 @@ async def create_ethernet_switch(project_id: UUID, node_data: schemas.EthernetSw """ builtin_manager = Builtin.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await builtin_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), - console=node_data.get("console"), - console_type=node_data.get("console_type"), + data.get("node_id"), + console=data.get("console"), + console_type=data.get("console_type"), node_type="ethernet_switch", - ports=node_data.get("ports_mapping"), + ports=data.get("ports_mapping"), ) - node.usage = node_data.get("usage", "") + node.usage = data.get("usage", "") return node.asdict() @@ -100,16 +103,16 @@ async def update_ethernet_switch( Update an Ethernet switch. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - if "name" in node_data and node.name != node_data["name"]: - node.name = node_data["name"] - if "usage" in node_data: - node.usage = node_data["usage"] - if "ports_mapping" in node_data: - node.ports_mapping = node_data["ports_mapping"] + data = jsonable_encoder(node_data, exclude_unset=True) + if "name" in data and node.name != data["name"]: + node.name = data["name"] + if "usage" in data: + node.usage = data["usage"] + if "ports_mapping" in data: + node.ports_mapping = data["ports_mapping"] await node.update_port_settings() - if "console_type" in node_data: - node.console_type = node_data["console_type"] + if "console_type" in data: + node.console_type = data["console_type"] node.updated() return node.asdict() diff --git a/gns3server/api/routes/compute/frame_relay_switch_nodes.py b/gns3server/api/routes/compute/frame_relay_switch_nodes.py index f425b3200..bd8bb1f18 100644 --- a/gns3server/api/routes/compute/frame_relay_switch_nodes.py +++ b/gns3server/api/routes/compute/frame_relay_switch_nodes.py @@ -18,6 +18,7 @@ API routes for Frame Relay switch nodes. """ +from typing import Any import os from fastapi import APIRouter, Depends, Body, Path, status, HTTPException @@ -29,7 +30,9 @@ from gns3server import schemas from gns3server.compute.dynamips import Dynamips from gns3server.compute.dynamips.nodes.frame_relay_switch import FrameRelaySwitch -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Frame Relay switch node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or Frame Relay switch node"} +} router = APIRouter(responses=responses) @@ -59,13 +62,13 @@ async def create_frame_relay_switch( # Use the Dynamips Frame Relay switch to simulate this node dynamips_manager = Dynamips.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await dynamips_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="frame_relay_switch", - mappings=node_data.get("mappings"), + mappings=data.get("mappings"), ) return node.asdict() @@ -99,11 +102,11 @@ async def update_frame_relay_switch( Update an Frame Relay switch node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - if "name" in node_data and node.name != node_data["name"]: - await node.set_name(node_data["name"]) - if "mappings" in node_data: - node.mappings = node_data["mappings"] + data = jsonable_encoder(node_data, exclude_unset=True) + if "name" in data and node.name != data["name"]: + await node.set_name(data["name"]) + if "mappings" in data: + node.mappings = data["mappings"] node.updated() return node.asdict() diff --git a/gns3server/api/routes/compute/iou_nodes.py b/gns3server/api/routes/compute/iou_nodes.py index 0282d0f0f..5285c07b5 100644 --- a/gns3server/api/routes/compute/iou_nodes.py +++ b/gns3server/api/routes/compute/iou_nodes.py @@ -23,7 +23,7 @@ import os from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse -from typing import Union +from typing import Any, Union from uuid import UUID from gns3server import schemas @@ -32,7 +32,9 @@ from gns3server.compute.iou.iou_vm import IOUVM from .dependencies.authentication import compute_authentication, ws_compute_authentication -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"} +} router = APIRouter(responses=responses) @@ -60,18 +62,18 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch """ iou = IOU.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) vm = await iou.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), - application_id=node_data.get("application_id"), - path=node_data.get("path"), - console=node_data.get("console"), - console_type=node_data.get("console_type", "telnet"), + data.get("node_id"), + application_id=data.get("application_id"), + path=data.get("path"), + console=data.get("console"), + console_type=data.get("console_type", "telnet"), ) - for name, value in node_data.items(): + for name, value in data.items(): if hasattr(vm, name) and getattr(vm, name) != value: if name == "application_id": continue # we must ignore this to avoid overwriting the application_id allocated by the controller @@ -79,7 +81,7 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch continue if name == "private_config_content" and (vm.private_config_content and len(vm.private_config_content) > 0): continue - if node_data.get("use_default_iou_values") and (name == "ram" or name == "nvram"): + if data.get("use_default_iou_values") and (name == "ram" or name == "nvram"): continue setattr(vm, name, value) return vm.asdict() @@ -100,8 +102,8 @@ async def update_iou_node(node_data: schemas.IOUUpdate, node: IOUVM = Depends(de Update an IOU node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - for name, value in node_data.items(): + data = jsonable_encoder(node_data, exclude_unset=True) + for name, value in data.items(): if hasattr(node, name) and getattr(node, name) != value: if name == "application_id": continue # we must ignore this to avoid overwriting the application_id allocated by the IOU manager @@ -147,8 +149,8 @@ async def start_iou_node(start_data: schemas.IOUStart, node: IOUVM = Depends(dep Start an IOU node. """ - start_data = jsonable_encoder(start_data, exclude_unset=True) - for name, value in start_data.items(): + start_values = jsonable_encoder(start_data, exclude_unset=True) + for name, value in start_values.items(): if hasattr(node, name) and getattr(node, name) != value: setattr(node, name, value) @@ -223,7 +225,7 @@ async def update_iou_node_nio( nio = node.get_nio(adapter_number, port_number) nio.filters.clear() - if nio_data.filters: + if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters: nio.filters = nio_data.filters # NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers. nio.markers = getattr(nio_data, "markers", None) or {} diff --git a/gns3server/api/routes/compute/nat_nodes.py b/gns3server/api/routes/compute/nat_nodes.py index 60654b603..37d8b4401 100644 --- a/gns3server/api/routes/compute/nat_nodes.py +++ b/gns3server/api/routes/compute/nat_nodes.py @@ -23,14 +23,16 @@ import os from fastapi import APIRouter, Depends, Path, status, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse -from typing import Union +from typing import Any, Union from uuid import UUID from gns3server import schemas from gns3server.compute.builtin import Builtin from gns3server.compute.builtin.nodes.nat import Nat -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or NAT node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or NAT node"} +} router = APIRouter(responses=responses) @@ -57,16 +59,16 @@ async def create_nat_node(project_id: UUID, node_data: schemas.NATCreate) -> sch """ builtin_manager = Builtin.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await builtin_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="nat", - ports=node_data.get("ports_mapping"), + ports=data.get("ports_mapping"), ) - node.usage = node_data.get("usage", "") + node.usage = data.get("usage", "") return node.asdict() @@ -85,8 +87,8 @@ async def update_nat_node(node_data: schemas.NATUpdate, node: Nat = Depends(dep_ Update a NAT node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - for name, value in node_data.items(): + data = jsonable_encoder(node_data, exclude_unset=True) + for name, value in data.items(): if hasattr(node, name) and getattr(node, name) != value: setattr(node, name, value) node.updated() @@ -170,7 +172,7 @@ async def update_nat_node_nio( nio = node.get_nio(port_number) nio.filters.clear() - if nio_data.filters: + if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters: nio.filters = nio_data.filters await node.update_nio(port_number, nio) return nio.asdict() diff --git a/gns3server/api/routes/compute/notifications.py b/gns3server/api/routes/compute/notifications.py index 2c6ca4ad5..ee533037b 100644 --- a/gns3server/api/routes/compute/notifications.py +++ b/gns3server/api/routes/compute/notifications.py @@ -39,14 +39,15 @@ async def project_ws_notifications(websocket: Union[None, WebSocket] = Depends(w """ if websocket: - log.info(f"New client {websocket.client.host}:{websocket.client.port} has connected to compute WebSocket") + client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown" + log.info(f"New client {client} has connected to compute WebSocket") try: with NotificationManager.instance().queue() as queue: while True: notification = await queue.get_json(5) await websocket.send_text(notification) except (ConnectionClosed, WebSocketDisconnect): - log.info(f"Client {websocket.client.host}:{websocket.client.port} has disconnected from compute WebSocket") + log.info(f"Client {client} has disconnected from compute WebSocket") except WebSocketException as e: log.warning(f"Error while sending to controller event to WebSocket client: {e}") diff --git a/gns3server/api/routes/controller/api_keys.py b/gns3server/api/routes/controller/api_keys.py index 235cbe7de..a3bfa4b14 100644 --- a/gns3server/api/routes/controller/api_keys.py +++ b/gns3server/api/routes/controller/api_keys.py @@ -39,7 +39,7 @@ API_KEY_PREFIX = "gns3_" API_KEY_BYTES = 32 -def _generate_api_key(api_key_id: UUID = None) -> tuple[str, str, str, UUID]: +def _generate_api_key(api_key_id: UUID | None = None) -> tuple[str, str, str, UUID]: if api_key_id is None: api_key_id = uuid4() random_bytes = secrets.token_hex(API_KEY_BYTES) diff --git a/gns3server/api/routes/controller/chat.py b/gns3server/api/routes/controller/chat.py index cc796fe26..e55e722f7 100644 --- a/gns3server/api/routes/controller/chat.py +++ b/gns3server/api/routes/controller/chat.py @@ -23,7 +23,7 @@ Nested under projects: /v3/projects/{project_id}/chat/... import json import logging import uuid -from typing import List, Optional +from typing import Any, List, Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.responses import StreamingResponse @@ -40,7 +40,7 @@ from .dependencies.authentication import get_current_active_user log = logging.getLogger(__name__) -responses = {404: {"model": schemas.ErrorMessage, "description": "Resource not found"}} +responses: dict[int | str, dict[str, Any]] = {404: {"model": schemas.ErrorMessage, "description": "Resource not found"}} router = APIRouter(responses=responses) diff --git a/gns3server/api/routes/controller/drawings.py b/gns3server/api/routes/controller/drawings.py index 42e989f0f..6df9ddd91 100644 --- a/gns3server/api/routes/controller/drawings.py +++ b/gns3server/api/routes/controller/drawings.py @@ -20,7 +20,7 @@ API routes for drawings. from fastapi import APIRouter, Depends, status from fastapi.encoders import jsonable_encoder -from typing import List +from typing import Any, List from uuid import UUID from gns3server.controller import Controller @@ -30,7 +30,9 @@ from gns3server import schemas from .dependencies.database import get_repository from .dependencies.rbac import has_privilege -responses = {404: {"model": schemas.ErrorMessage, "description": "Project or drawing not found"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Project or drawing not found"} +} router = APIRouter(responses=responses) diff --git a/gns3server/api/routes/controller/nodes.py b/gns3server/api/routes/controller/nodes.py index 0938ab087..8cd7c0ade 100644 --- a/gns3server/api/routes/controller/nodes.py +++ b/gns3server/api/routes/controller/nodes.py @@ -27,7 +27,7 @@ from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, Request, from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse from fastapi.routing import APIRoute -from typing import List, Callable, Optional +from typing import Any, List, Callable, Optional from uuid import UUID from gns3server.controller import Controller @@ -47,7 +47,7 @@ import logging log = logging.getLogger(__name__) -node_locks = {} +node_locks: dict[str, dict[str, Any]] = {} class NodeConcurrency(APIRoute): @@ -84,7 +84,9 @@ class NodeConcurrency(APIRoute): return custom_route_handler -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or node"} +} router = APIRouter(route_class=NodeConcurrency, responses=responses) @@ -127,7 +129,7 @@ def _check_node_type(node: Node, *required_types: str) -> None: }, dependencies=[Depends(has_privilege("Node.Allocate"))], ) -async def create_node(node_data: schemas.NodeCreate, project: Project = Depends(dep_project)) -> schemas.Node: +async def create_node(node_create: schemas.NodeCreate, project: Project = Depends(dep_project)) -> schemas.Node: """ Create a new node. @@ -135,8 +137,8 @@ async def create_node(node_data: schemas.NodeCreate, project: Project = Depends( """ controller = Controller.instance() - compute = controller.get_compute(str(node_data.compute_id)) - node_data = jsonable_encoder(node_data, exclude_unset=True) + compute = controller.get_compute(str(node_create.compute_id)) + node_data = jsonable_encoder(node_create, exclude_unset=True) node = await project.add_node(compute, node_data.pop("name"), node_data.pop("node_id", None), **node_data) return node.asdict() @@ -274,14 +276,14 @@ async def get_node(node: Node = Depends(dep_node)) -> schemas.Node: response_model_exclude_unset=True, dependencies=[Depends(has_privilege("Node.Modify"))], ) -async def update_node(node_data: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node: +async def update_node(node_update: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node: """ Update a node. Required privilege: Node.Modify """ - node_data = jsonable_encoder(node_data, exclude_unset=True) + node_data = jsonable_encoder(node_update, exclude_unset=True) # Ignore these because we only use them when creating a node node_data.pop("node_id", None) @@ -654,9 +656,8 @@ async def ws_console( return compute = node.compute - log.info( - f"New client {websocket.client.host}:{websocket.client.port} has connected to controller console WebSocket" - ) + client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown" + log.info(f"New client {client} has connected to controller console WebSocket") compute_host = compute.host try: @@ -690,9 +691,7 @@ async def ws_console( await ws_console_compute.send_bytes(msg["bytes"]) except WebSocketDisconnect: pass - log.info( - f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller console WebSocket" - ) + log.info(f"Client {client} has disconnected from controller console WebSocket") async def ws_send(ws_console_compute): """ @@ -709,10 +708,7 @@ async def ws_console( break except WebSocketDisconnect: # the client disconnected while the compute was still streaming console output - log.info( - f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller" - f" console WebSocket" - ) + log.info(f"Client {client} has disconnected from controller console WebSocket") try: # forward WebSocket data in both directions between the client and the compute console WebSocket @@ -763,9 +759,8 @@ async def vnc_console( return compute = node.compute - log.info( - f"New client {websocket.client.host}:{websocket.client.port} has connected to controller VNC console WebSocket" - ) + client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown" + log.info(f"New client {client} has connected to controller VNC console WebSocket") compute_host = compute.host try: @@ -796,10 +791,7 @@ async def vnc_console( await vnc_console_compute.send_bytes(data) except WebSocketDisconnect: pass - log.info( - f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller" - f" VNC console WebSocket" - ) + log.info(f"Client {client} has disconnected from controller VNC console WebSocket") async def vnc_send(vnc_console_compute): """ @@ -814,10 +806,7 @@ async def vnc_console( break except WebSocketDisconnect: # the client disconnected while the compute was still streaming VNC console output - log.info( - f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller" - f" VNC console WebSocket" - ) + log.info(f"Client {client} has disconnected from controller VNC console WebSocket") try: # forward WebSocket data in both directions between the client and the compute VNC console WebSocket diff --git a/gns3server/api/routes/controller/privileges.py b/gns3server/api/routes/controller/privileges.py index a070a6ee6..1f34ec1ff 100644 --- a/gns3server/api/routes/controller/privileges.py +++ b/gns3server/api/routes/controller/privileges.py @@ -16,6 +16,7 @@ API route for privileges """ from typing import List +import gns3server.db.models as models from gns3server.db.repositories.rbac import RbacRepository from .dependencies.database import get_repository from fastapi import APIRouter, Depends @@ -33,7 +34,7 @@ router = APIRouter() ) async def get_privileges( rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), -) -> List[schemas.Privilege]: +) -> List[models.Privilege]: """ Get all privileges. diff --git a/gns3server/api/routes/controller/roles.py b/gns3server/api/routes/controller/roles.py index aab6e5a00..3f6e854a1 100644 --- a/gns3server/api/routes/controller/roles.py +++ b/gns3server/api/routes/controller/roles.py @@ -21,7 +21,7 @@ API routes for roles. from fastapi import APIRouter, Depends, status from uuid import UUID -from typing import List +from typing import List, Optional from gns3server import schemas from gns3server.controller.controller_error import ( @@ -31,6 +31,7 @@ from gns3server.controller.controller_error import ( ControllerForbiddenError, ) +import gns3server.db.models as models from gns3server.db.repositories.rbac import RbacRepository from .dependencies.database import get_repository from .dependencies.rbac import has_privilege @@ -43,7 +44,7 @@ router = APIRouter() @router.get("", response_model=List[schemas.Role], dependencies=[Depends(has_privilege("Role.Audit"))]) -async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[schemas.Role]: +async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[models.Role]: """ Get all roles. @@ -61,7 +62,7 @@ async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepos ) async def create_role( role_create: schemas.RoleCreate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)) -) -> schemas.Role: +) -> Optional[models.Role]: """ Create a new role. @@ -78,7 +79,7 @@ async def create_role( async def get_role( role_id: UUID, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), -) -> schemas.Role: +) -> models.Role: """ Get a role. @@ -94,7 +95,7 @@ async def get_role( @router.put("/{role_id}", response_model=schemas.Role, dependencies=[Depends(has_privilege("Role.Modify"))]) async def update_role( role_id: UUID, role_update: schemas.RoleUpdate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)) -) -> schemas.Role: +) -> Optional[models.Role]: """ Update a role. @@ -142,7 +143,7 @@ async def delete_role( ) async def get_role_privileges( role_id: UUID, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)) -) -> List[schemas.Privilege]: +) -> List[models.Privilege]: """ Get all role privileges. diff --git a/gns3server/api/routes/controller/snapshots.py b/gns3server/api/routes/controller/snapshots.py index f85b5d9ef..4ec8c467c 100644 --- a/gns3server/api/routes/controller/snapshots.py +++ b/gns3server/api/routes/controller/snapshots.py @@ -24,7 +24,7 @@ import logging log = logging.getLogger() from fastapi import APIRouter, Depends, status -from typing import List +from typing import Any, List from uuid import UUID from gns3server.controller.project import Project @@ -35,7 +35,9 @@ from gns3server.controller import Controller from .dependencies.database import get_repository from .dependencies.rbac import has_privilege -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or snapshot"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or snapshot"} +} router = APIRouter(responses=responses) diff --git a/gns3server/controller/topology.py b/gns3server/controller/topology.py index 5048be4ce..f2390930a 100644 --- a/gns3server/controller/topology.py +++ b/gns3server/controller/topology.py @@ -53,7 +53,7 @@ GNS3_FILE_FORMAT_REVISION = 10 class DynamipsNodeValidation(DynamipsCreate): - name: Optional[str] = None + name: Optional[str] = None # type: ignore[assignment] def _check_topology_schema(topo, path): diff --git a/gns3server/schemas/compute/atm_switch_nodes.py b/gns3server/schemas/compute/atm_switch_nodes.py index 9a9db0845..9d126bfe9 100644 --- a/gns3server/schemas/compute/atm_switch_nodes.py +++ b/gns3server/schemas/compute/atm_switch_nodes.py @@ -26,8 +26,8 @@ class ATMSwitchBase(BaseModel): Common ATM switch properties. """ - name: str = None - node_id: UUID = None + name: Optional[str] = None + node_id: Optional[UUID] = None usage: Optional[str] = None mappings: Optional[dict] = None diff --git a/gns3server/schemas/compute/dynamips_nodes.py b/gns3server/schemas/compute/dynamips_nodes.py index d01020a28..6292b400b 100644 --- a/gns3server/schemas/compute/dynamips_nodes.py +++ b/gns3server/schemas/compute/dynamips_nodes.py @@ -124,7 +124,6 @@ class DynamipsBase(BaseModel): node_id: Optional[UUID] = None name: Optional[str] = None dynamips_id: Optional[int] = Field(None, description="Dynamips internal ID") - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") ram: Optional[int] = Field(None, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(None, gt=0, description="Amount of NVRAM in KB") image: Optional[str] = Field(None, description="Path to the IOS image") @@ -185,11 +184,12 @@ class DynamipsUpdate(DynamipsBase): Properties to update a Dynamips node. """ - pass + platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") class Dynamips(DynamipsBase): name: str + platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") node_id: UUID project_id: UUID dynamips_id: int diff --git a/gns3server/schemas/compute/nat_nodes.py b/gns3server/schemas/compute/nat_nodes.py index 95a703320..e3c569251 100644 --- a/gns3server/schemas/compute/nat_nodes.py +++ b/gns3server/schemas/compute/nat_nodes.py @@ -89,7 +89,7 @@ class NATBase(BaseModel): Common NAT node properties. """ - name: str + name: Optional[str] = None node_id: Optional[UUID] = None usage: Optional[str] = None ports_mapping: Optional[List[Union[EthernetPort, TAPPort, UDPPort]]] = Field( @@ -102,7 +102,7 @@ class NATCreate(NATBase): Properties to create a NAT node. """ - pass + name: str class NATUpdate(NATBase): @@ -110,10 +110,11 @@ class NATUpdate(NATBase): Properties to update a NAT node. """ - name: Optional[str] = None + pass class NAT(NATBase): + name: str project_id: UUID node_id: UUID ports_mapping: List[Union[EthernetPort, TAPPort, UDPPort]] diff --git a/gns3server/utils/encryption.py b/gns3server/utils/encryption.py index 1f95f497b..38719a60e 100644 --- a/gns3server/utils/encryption.py +++ b/gns3server/utils/encryption.py @@ -48,7 +48,7 @@ def init_encryption(secrets_dir: str) -> None: os.makedirs(secrets_dir, exist_ok=True) with open(encryption_key_path, "w", encoding="utf-8") as f: # Use Fernet's base64-encoded key format - f.write(key.decode() if isinstance(key, bytes) else key) + f.write(key.decode()) # Set restrictive permissions (owner read/write only) os.chmod(encryption_key_path, 0o600) except OSError as e: @@ -58,7 +58,7 @@ def init_encryption(secrets_dir: str) -> None: try: with open(encryption_key_path, encoding="utf-8") as f: key_content = f.read().strip() - key_bytes = key_content.encode() if isinstance(key_content, str) else key_content + key_bytes = key_content.encode() _fernet = Fernet(key_bytes) log.debug("Encryption initialized successfully") except OSError as e: diff --git a/gns3server/utils/images.py b/gns3server/utils/images.py index c06e22bec..06b4ae9be 100644 --- a/gns3server/utils/images.py +++ b/gns3server/utils/images.py @@ -126,7 +126,7 @@ def get_builtin_disks() -> List[str]: return builtin_disks -async def read_image_info(path: str, expected_image_type: str = None) -> dict: +async def read_image_info(path: str, expected_image_type: str | None = None) -> dict: header_magic_len = 7 try: @@ -155,7 +155,7 @@ async def read_image_info(path: str, expected_image_type: str = None) -> dict: return image_info -async def discover_images(image_type: str, skip_image_paths: list = None) -> List[dict]: +async def discover_images(image_type: str, skip_image_paths: list | None = None) -> List[dict]: """ Scan directories for available images """ @@ -378,9 +378,9 @@ async def write_image( f"File '{image_path}' already exists, please choose a different name or remove the existing image" ) - checksum = checksum.hexdigest() + checksum_hex = checksum.hexdigest() image_dir = os.path.dirname(image_path) - duplicate_image = await images_repo.get_image_by_checksum(checksum, image_dir) + duplicate_image = await images_repo.get_image_by_checksum(checksum_hex, image_dir) if duplicate_image: raise InvalidImageError( f"Image '{duplicate_image.filename}' with the same checksum already exists in '{image_dir}'" @@ -396,5 +396,5 @@ async def write_image( log.warning(f"Could not remove '{tmp_path}'") return await images_repo.add_image( - image_name, image_type, image_size, image_path, checksum, checksum_algorithm="md5" + image_name, image_type, image_size, image_path, checksum_hex, checksum_algorithm="md5" ) diff --git a/gns3server/utils/packet_filter_validation.py b/gns3server/utils/packet_filter_validation.py index cc2634546..1b9481ffd 100644 --- a/gns3server/utils/packet_filter_validation.py +++ b/gns3server/utils/packet_filter_validation.py @@ -15,7 +15,7 @@ class FilterValidationError(Exception): pass -def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Optional[str]]: +def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Any]: """ Validate BPF filter expression syntax using tcpdump. @@ -78,7 +78,7 @@ def validate_filter_parameters(filter_type: str, values: List[Any]) -> None: """ # Define validation rules based on ubridge implementation - VALIDATION_RULES = { + VALIDATION_RULES: Dict[str, Dict[str, Any]] = { "frequency_drop": { "params_count": 1, "ranges": [(-1, 32767)], # min, max @@ -173,11 +173,11 @@ def filter_inactive_filters(filters: Dict[str, List[Any]]) -> Dict[str, List[Any active_filters = {} for filter_type, values in filters.items(): - if not values or (isinstance(values, list) and len(values) == 0): + if not values: continue # Normalize values (strip strings, convert to int) - normalized_values = [] + normalized_values: List[Any] = [] for value in values: if isinstance(value, str): normalized_values.append(value.strip("\n ")) @@ -228,7 +228,7 @@ def validate_all_filters(filters: Dict[str, List[Any]]) -> None: return for filter_type, values in filters.items(): - if not values or (isinstance(values, list) and len(values) == 0): + if not values: continue validate_filter_parameters(filter_type, values) diff --git a/pyproject.toml b/pyproject.toml index 5a022c067..bb1f95c75 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -294,63 +294,18 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"] # Baseline: modules with existing type errors (error count at baseline time). # Remove modules from this list one small PR at a time. Never add new ones. module = [ - "gns3server.agent", # 2 - "gns3server.agent.gns3_copilot.agent.context_manager", # 5 - "gns3server.agent.gns3_copilot.agent.model_factory", # 4 - "gns3server.agent.gns3_copilot.chat_sessions_repository", # 8 - "gns3server.agent.gns3_copilot.gns3_client.api_handlers", # 10 - "gns3server.agent.gns3_copilot.gns3_client.connector_factory", # 1 - "gns3server.agent.gns3_copilot.gns3_client.project_inventory", # 2 - "gns3server.agent.gns3_copilot.skills.loader", # 1 - "gns3server.agent.gns3_copilot.skills.manager", # 4 - "gns3server.agent.gns3_copilot.tools_v2.config_tools_nornir", # 6 - "gns3server.agent.gns3_copilot.tools_v2.display_tools_nornir", # 6 - "gns3server.agent.gns3_copilot.tools_v2.gns3_packet_filter", # 1 - "gns3server.agent.gns3_copilot.tools_v2.gns3_start_node", # 3 - "gns3server.agent.gns3_copilot.tools_v2.gns3_stop_node", # 3 - "gns3server.agent.gns3_copilot.tools_v2.gns3_suspend_node", # 3 - "gns3server.agent.gns3_copilot.tools_v2.packet_analysis_tool", # 1 - "gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko", # 9 - "gns3server.agent.gns3_copilot.utils.custom_netmiko.scripts.list_netmiko_telnet_devices", # 1 - "gns3server.agent.gns3_copilot.utils.message_converters", # 2 - "gns3server.agent.gns3_copilot.utils.parse_tool_content", # 1 - "gns3server.agent.mcp.appliances", # 1 - "gns3server.agent.web_wireshark.docker_client", # 5 - "gns3server.agent.web_wireshark.manage_wireshark", # 1 - "gns3server.agent.web_wireshark.manager", # 9 - "gns3server.agent.web_wireshark.stats", # 4 - "gns3server.api.routes.compute.atm_switch_nodes", # 9 - "gns3server.api.routes.compute.capabilities", # 1 - "gns3server.api.routes.compute.cloud_nodes", # 14 - "gns3server.api.routes.compute.compute", # 1 - "gns3server.api.routes.compute.dependencies.authentication", # 3 - "gns3server.api.routes.compute.docker_nodes", # 26 - "gns3server.api.routes.compute.dynamips_nodes", # 9 - "gns3server.api.routes.compute.ethernet_hub_nodes", # 9 - "gns3server.api.routes.compute.ethernet_switch_nodes", # 16 - "gns3server.api.routes.compute.frame_relay_switch_nodes", # 9 - "gns3server.api.routes.compute.iou_nodes", # 15 - "gns3server.api.routes.compute.nat_nodes", # 10 - "gns3server.api.routes.compute.notifications", # 4 "gns3server.api.routes.compute.vmware_nodes", # 10 "gns3server.api.routes.compute.vpcs_nodes", # 9 "gns3server.api.routes.controller.acl", # 5 - "gns3server.api.routes.controller.api_keys", # 1 - "gns3server.api.routes.controller.chat", # 1 "gns3server.api.routes.controller.computes", # 6 "gns3server.api.routes.controller.controller", # 10 "gns3server.api.routes.controller.dependencies.authentication", # 11 "gns3server.api.routes.controller.dependencies.database", # 2 - "gns3server.api.routes.controller.drawings", # 1 "gns3server.api.routes.controller.groups", # 6 "gns3server.api.routes.controller.images", # 9 "gns3server.api.routes.controller.llm_model_configs", # 32 - "gns3server.api.routes.controller.nodes", # 13 "gns3server.api.routes.controller.pools", # 8 - "gns3server.api.routes.controller.privileges", # 1 "gns3server.api.routes.controller.projects", # 16 - "gns3server.api.routes.controller.roles", # 5 - "gns3server.api.routes.controller.snapshots", # 1 "gns3server.api.routes.controller.templates", # 10 "gns3server.api.routes.controller.users", # 14 "gns3server.api.server", # 2 @@ -362,7 +317,6 @@ module = [ "gns3server.controller.appliance", # 2 "gns3server.controller.appliance_manager", # 5 "gns3server.controller.project", # 2 - "gns3server.controller.topology", # 1 "gns3server.db.repositories.api_keys", # 8 "gns3server.db.repositories.computes", # 10 "gns3server.db.repositories.images", # 14 @@ -372,13 +326,10 @@ module = [ "gns3server.db.repositories.templates", # 6 "gns3server.db.repositories.users", # 12 "gns3server.db.tasks", # 12 - "gns3server.schemas.compute.atm_switch_nodes", # 5 "gns3server.schemas.compute.cloud_nodes", # 1 "gns3server.schemas.compute.docker_nodes", # 2 - "gns3server.schemas.compute.dynamips_nodes", # 1 "gns3server.schemas.compute.frame_relay_switch_nodes", # 5 "gns3server.schemas.compute.iou_nodes", # 3 - "gns3server.schemas.compute.nat_nodes", # 1 "gns3server.schemas.compute.qemu_nodes", # 1 "gns3server.schemas.compute.virtualbox_nodes", # 2 "gns3server.schemas.compute.vmware_nodes", # 3 @@ -396,10 +347,7 @@ module = [ "gns3server.services.computes", # 4 "gns3server.services.templates", # 18 "gns3server.utils.asyncio.aiozipstream", # 8 - "gns3server.utils.encryption", # 2 "gns3server.utils.http_client", # 4 - "gns3server.utils.images", # 4 - "gns3server.utils.packet_filter_validation", # 16 "gns3server.utils.websocket_to_websocket", # 12 "gns3server.utils.zipfile_zstd._patcher", # 1 ]