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/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py index bf916b043..32c71c981 100644 --- a/gns3server/agent/gns3_copilot/agent_service.py +++ b/gns3server/agent/gns3_copilot/agent_service.py @@ -256,17 +256,16 @@ class AgentService: ) # Ensure checkpointer is initialized - if not self._checkpointer_conn: - await self._get_checkpointer() + checkpointer = await self._get_checkpointer() # Get or create chat session - repo = ChatSessionsRepository(self._checkpointer_conn) + repo = ChatSessionsRepository(checkpointer.conn) session = await repo.get_session_by_thread(session_id) is_new_session = session is None if is_new_session: # Create new session - copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower() + copilot_mode = (llm_config or {}).get("copilot_mode", "teaching_assistant").lower() session = await repo.create_session( thread_id=session_id, user_id=user_id or "", @@ -290,7 +289,7 @@ class AgentService: # Build config - only thread-safe identifiers # Determine recursion_limit based on copilot_mode - copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower() + copilot_mode = (llm_config or {}).get("copilot_mode", "teaching_assistant").lower() if copilot_mode == "troubleshooting_injection": recursion_limit = 100 # Need more recursion depth for fault injection workflow log.debug("Using extended recursion_limit for troubleshooting_injection mode: 100") @@ -453,10 +452,10 @@ class AgentService: yield chunk else: # Use stateless converter for other events - chunk = self._convert_event_to_chunk(event, session_id) - if chunk: - log.debug("Yielding chunk: type=%s", chunk.get("type")) - yield chunk + converted = self._convert_event_to_chunk(event, session_id) + if converted: + log.debug("Yielding chunk: type=%s", converted.get("type")) + yield converted # Check if stream was aborted and yield tool_end events for aborted tools from gns3server.agent.gns3_copilot.agent.gns3_copilot import ( @@ -624,10 +623,9 @@ class AgentService: Returns: List of session dictionaries """ - if not self._checkpointer_conn: - await self._get_checkpointer() + checkpointer = await self._get_checkpointer() - repo = ChatSessionsRepository(self._checkpointer_conn) + repo = ChatSessionsRepository(checkpointer.conn) sessions = await repo.list_sessions(user_id=user_id, copilot_mode=copilot_mode, limit=limit) return [s.to_dict() for s in sessions] @@ -641,10 +639,9 @@ class AgentService: Returns: True if deleted, False if not found """ - if not self._checkpointer_conn: - await self._get_checkpointer() + checkpointer = await self._get_checkpointer() - repo = ChatSessionsRepository(self._checkpointer_conn) + repo = ChatSessionsRepository(checkpointer.conn) return await repo.delete_session(session_id) async def rename_session(self, session_id: str, new_title: str) -> Optional[Dict[str, Any]]: @@ -658,10 +655,9 @@ class AgentService: Returns: Updated session dictionary or None """ - if not self._checkpointer_conn: - await self._get_checkpointer() + checkpointer = await self._get_checkpointer() - repo = ChatSessionsRepository(self._checkpointer_conn) + repo = ChatSessionsRepository(checkpointer.conn) session = await repo.update_session(thread_id=session_id, title=new_title) return session.to_dict() if session else None @@ -676,10 +672,9 @@ class AgentService: Returns: Updated session dictionary or None """ - if not self._checkpointer_conn: - await self._get_checkpointer() + checkpointer = await self._get_checkpointer() - repo = ChatSessionsRepository(self._checkpointer_conn) + repo = ChatSessionsRepository(checkpointer.conn) session = await repo.pin_session(thread_id=session_id, pinned=pinned) return session.to_dict() if session else None 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/mcp/__init__.py b/gns3server/agent/mcp/__init__.py index 9b9026253..e89babc77 100644 --- a/gns3server/agent/mcp/__init__.py +++ b/gns3server/agent/mcp/__init__.py @@ -405,22 +405,24 @@ async def project_stats( @mcp.tool() async def project_update( project_id: Annotated[str, Field(description="UUID of the project to update")], - name: Annotated[str, Field(description="New project name")] = None, - auto_close: Annotated[bool, Field(description="Close project when last client leaves")] = None, - auto_open: Annotated[bool, Field(description="Project opens when GNS3 starts")] = None, - auto_start: Annotated[bool, Field(description="Project starts when opened")] = None, - scene_width: Annotated[int, Field(description="Width of the drawing area")] = None, - scene_height: Annotated[int, Field(description="Height of the drawing area")] = None, - zoom: Annotated[int, Field(description="Zoom of the drawing area")] = None, - show_layers: Annotated[bool, Field(description="Show layers on the drawing area")] = None, - snap_to_grid: Annotated[bool, Field(description="Snap to grid on the drawing area")] = None, - show_grid: Annotated[bool, Field(description="Show the grid on the drawing area")] = None, - grid_size: Annotated[int, Field(description="Grid size for the drawing area for nodes")] = None, - drawing_grid_size: Annotated[int, Field(description="Grid size for the drawing area for drawings")] = None, - show_interface_labels: Annotated[bool, Field(description="Show interface labels on the drawing area")] = None, + name: Annotated[str | None, Field(description="New project name")] = None, + auto_close: Annotated[bool | None, Field(description="Close project when last client leaves")] = None, + auto_open: Annotated[bool | None, Field(description="Project opens when GNS3 starts")] = None, + auto_start: Annotated[bool | None, Field(description="Project starts when opened")] = None, + scene_width: Annotated[int | None, Field(description="Width of the drawing area")] = None, + scene_height: Annotated[int | None, Field(description="Height of the drawing area")] = None, + zoom: Annotated[int | None, Field(description="Zoom of the drawing area")] = None, + show_layers: Annotated[bool | None, Field(description="Show layers on the drawing area")] = None, + snap_to_grid: Annotated[bool | None, Field(description="Snap to grid on the drawing area")] = None, + show_grid: Annotated[bool | None, Field(description="Show the grid on the drawing area")] = None, + grid_size: Annotated[int | None, Field(description="Grid size for the drawing area for nodes")] = None, + drawing_grid_size: Annotated[int | None, Field(description="Grid size for the drawing area for drawings")] = None, + show_interface_labels: Annotated[ + bool | None, Field(description="Show interface labels on the drawing area") + ] = None, ) -> list[dict[str, Any]]: """Update a project's properties (name, auto_close, auto_open, etc.).""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} local_vars = { "name": name, "auto_close": auto_close, @@ -449,7 +451,7 @@ async def project_duplicate( reset_mac_addresses: Annotated[bool, Field(description="Reset MAC addresses for this project")] = False, ) -> list[dict[str, Any]]: """Duplicate a project.""" - params = {"project_id": project_id, "name": name} + params: dict[str, Any] = {"project_id": project_id, "name": name} if reset_mac_addresses: params["reset_mac_addresses"] = reset_mac_addresses return await asyncio.to_thread(_run_handler_sync, duplicate_project_handler, params) @@ -523,7 +525,7 @@ async def node_start( ] = None, ) -> list[dict[str, Any]]: """Start one or more nodes. Provide node_id for single, or node_ids for batch.""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if node_ids: params["node_ids"] = node_ids else: @@ -540,7 +542,7 @@ async def node_stop( ] = None, ) -> list[dict[str, Any]]: """Stop one or more nodes. Provide node_id for single, or node_ids for batch.""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if node_ids: params["node_ids"] = node_ids else: @@ -557,7 +559,7 @@ async def node_suspend( ] = None, ) -> list[dict[str, Any]]: """Suspend one or more nodes. Provide node_id for single, or node_ids for batch.""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if node_ids: params["node_ids"] = node_ids else: @@ -635,7 +637,7 @@ async def node_delete( ] = None, ) -> list[dict[str, Any]]: """Delete one or more nodes from a project. Provide node_id for single, or node_ids for batch.""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if node_ids: params["node_ids"] = node_ids else: @@ -756,7 +758,7 @@ async def link_create( "fields": fields, }, ) - params = {"project_id": project_id, "nodes": nodes, "link_type": link_type, "fields": fields} + params: dict[str, Any] = {"project_id": project_id, "nodes": nodes, "link_type": link_type, "fields": fields} if filters: params["filters"] = filters return await asyncio.to_thread(_run_handler_sync, create_link_handler, params) @@ -771,7 +773,7 @@ async def link_delete( ] = None, ) -> list[dict[str, Any]]: """Delete one or more links from a project.""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if link_ids: params["link_ids"] = link_ids else: @@ -1201,7 +1203,7 @@ async def link_reset( while keeping the filter configuration intact. Filters are preserved but their internal application state resets. """ - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if link_ids: params["link_ids"] = link_ids else: @@ -1222,7 +1224,7 @@ async def link_capture_start( ] = None, ) -> list[dict[str, Any]]: """Start packet capture on one or more links.""" - params = { + params: dict[str, Any] = { "project_id": project_id, "data_link_type": data_link_type, "capture_file_name": capture_file_name, @@ -1245,7 +1247,7 @@ async def link_capture_stop( ] = None, ) -> list[dict[str, Any]]: """Stop packet capture on one or more links.""" - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if link_ids: params["link_ids"] = link_ids else: @@ -1273,7 +1275,7 @@ async def link_capture_download( - The ticket expires after 10 minutes: call this tool again to get a fresh one; do not reuse an old URL. """ - params = {"project_id": project_id} + params: dict[str, Any] = {"project_id": project_id} if link_ids: params["link_ids"] = link_ids else: @@ -1555,7 +1557,7 @@ async def drawing_update( rotation: Annotated[int | None, Field(description="Rotation angle in degrees, -359 to 359")] = None, ) -> list[dict[str, Any]]: """Update a drawing's properties (svg, position, lock state, rotation, etc.).""" - params = {"project_id": project_id, "drawing_id": drawing_id} + params: dict[str, Any] = {"project_id": project_id, "drawing_id": drawing_id} local_vars = {"svg": svg, "locked": locked, "x": x, "y": y, "z": z, "rotation": rotation} for key, val in local_vars.items(): if val is not None: 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/qemu_nodes.py b/gns3server/api/routes/compute/qemu_nodes.py index 4f8498847..6f1794fb9 100644 --- a/gns3server/api/routes/compute/qemu_nodes.py +++ b/gns3server/api/routes/compute/qemu_nodes.py @@ -23,7 +23,7 @@ import os from fastapi import APIRouter, WebSocket, Depends, Body, 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 @@ -36,7 +36,9 @@ import logging log = logging.getLogger(__name__) -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Qemu node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or Qemu node"} +} router = APIRouter(responses=responses) @@ -64,29 +66,29 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s """ qemu = Qemu.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) - disk_images_to_reset = set(node_data.pop("disk_images_to_reset", [])) + data = jsonable_encoder(node_data, exclude_unset=True) + disk_images_to_reset = set(data.pop("disk_images_to_reset", [])) vm = await qemu.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.pop("node_id", None), - linked_clone=node_data.get("linked_clone", True), - qemu_path=node_data.pop("qemu_path", None), - console=node_data.pop("console", None), - console_type=node_data.pop("console_type", "telnet"), - aux=node_data.get("aux"), - aux_type=node_data.pop("aux_type", "none"), - platform=node_data.pop("platform", None), + data.pop("node_id", None), + linked_clone=data.get("linked_clone", True), + qemu_path=data.pop("qemu_path", None), + console=data.pop("console", None), + console_type=data.pop("console_type", "telnet"), + aux=data.get("aux"), + aux_type=data.pop("aux_type", "none"), + platform=data.pop("platform", None), ) # update the disk image with the backing file if provided # this is needed when duplicating a node that uses backed disk images drives = ["a", "b", "c", "d"] for drive in drives: - disk_image_backing_file = node_data.get(f"hd{drive}_disk_image_backing_file") + disk_image_backing_file = data.get(f"hd{drive}_disk_image_backing_file") if disk_image_backing_file: log.debug(f"Updating disk image for drive {drive} with backing file {disk_image_backing_file}") - node_data[f"hd{drive}_disk_image"] = disk_image_backing_file + data[f"hd{drive}_disk_image"] = disk_image_backing_file # Validate every explicitly replaced disk before removing its stale # overlay. Other unresolved disks may still make this create request fail, @@ -94,7 +96,7 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s for drive in drives: disk_image_property = f"hd{drive}_disk_image" if disk_image_property in disk_images_to_reset: - replacement_image = node_data.get(disk_image_property) + replacement_image = data.get(disk_image_property) if replacement_image: vm.manager.get_abs_image_path(replacement_image, vm.working_dir) local_disk_name = f"hd{drive}_disk.qcow2" @@ -107,11 +109,11 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s log.info( "Removing stale linked-clone disk '%s' before using replacement image '%s'", local_disk, - node_data.get(disk_image_property), + data.get(disk_image_property), ) vm.delete_disk_image(local_disk_name) - for name, value in node_data.items(): + for name, value in data.items(): if hasattr(vm, name) and getattr(vm, name) != value: setattr(vm, name, value) @@ -133,10 +135,10 @@ async def update_qemu_node(node_data: schemas.QemuUpdate, node: QemuVM = Depends Update a Qemu node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) # update the console first to avoid issue if updating console type - node.console = node_data.pop("console", node.console) - for name, value in node_data.items(): + node.console = data.pop("console", node.console) + for name, value in data.items(): if hasattr(node, name) and getattr(node, name) != value: await node.update_property(name, value) node.updated() diff --git a/gns3server/api/routes/controller/links.py b/gns3server/api/routes/controller/links.py index 1ee60a891..2f5eddee3 100644 --- a/gns3server/api/routes/controller/links.py +++ b/gns3server/api/routes/controller/links.py @@ -26,7 +26,7 @@ import aiohttp from fastapi import APIRouter, Depends, Request, status, WebSocket from fastapi.responses import FileResponse, StreamingResponse from fastapi.encoders import jsonable_encoder -from typing import List, Union +from typing import Any, List, Union from uuid import UUID, uuid4 from gns3server.controller import Controller @@ -46,7 +46,9 @@ import logging log = logging.getLogger(__name__) -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or link"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or link"} +} router = APIRouter(responses=responses) @@ -91,7 +93,7 @@ async def get_links(project_id: UUID) -> List[schemas.Link]: }, dependencies=[Depends(has_privilege("Link.Allocate"))], ) -async def create_link(project_id: UUID, link_data: schemas.LinkCreate) -> schemas.Link: +async def create_link(project_id: UUID, link_create: schemas.LinkCreate) -> schemas.Link: """ Create a new link. @@ -100,7 +102,7 @@ async def create_link(project_id: UUID, link_data: schemas.LinkCreate) -> schema project = await Controller.instance().get_loaded_project(str(project_id)) link = await project.add_link() - link_data = jsonable_encoder(link_data, exclude_unset=True) + link_data = jsonable_encoder(link_create, exclude_unset=True) if "filters" in link_data: await link.update_filters(link_data["filters"]) if "link_style" in link_data: @@ -156,14 +158,14 @@ async def get_link(link: Link = Depends(dep_link)) -> schemas.Link: response_model_exclude_unset=True, dependencies=[Depends(has_privilege("Link.Modify"))], ) -async def update_link(link_data: schemas.LinkUpdate, link: Link = Depends(dep_link)) -> schemas.Link: +async def update_link(link_update: schemas.LinkUpdate, link: Link = Depends(dep_link)) -> schemas.Link: """ Update a link. Required privilege: Link.Modify """ - link_data = jsonable_encoder(link_data, exclude_unset=True) + link_data = jsonable_encoder(link_update, exclude_unset=True) if "filters" in link_data: await link.update_filters(link_data["filters"]) if "link_style" in link_data: @@ -293,7 +295,7 @@ async def stream_pcap(request: Request, link: Link = Depends(dep_link)) -> Strea pcap_streaming_url = link.pcap_streaming_url() headers = multidict.MultiDict(request.headers) headers["Host"] = compute.host - headers["Router-Host"] = request.client.host + headers["Router-Host"] = request.client.host if request.client else "" body = await request.body() async def compute_pcap_stream(): @@ -487,7 +489,7 @@ async def update_marker(marker_name: str, marker_data: schemas.MarkerUpdate, lin response_model=Union[schemas.UDPPortInfo, schemas.EthernetPortInfo], dependencies=[Depends(has_privilege("Link.Audit"))], ) -async def get_iface(link: Link = Depends(dep_link)) -> Union[schemas.UDPPortInfo, schemas.EthernetPortInfo]: +async def get_iface(link: Link = Depends(dep_link)) -> dict: """ Return iface info for links to Cloud or NAT devices. diff --git a/gns3server/compute/iou/iou_vm.py b/gns3server/compute/iou/iou_vm.py index 4e329aad5..b0f99a4c2 100644 --- a/gns3server/compute/iou/iou_vm.py +++ b/gns3server/compute/iou/iou_vm.py @@ -139,8 +139,8 @@ class IOUVM(BaseNode): # Class-level caches shared across all IOU VM instances using the same image. # These avoid redundant subprocess calls during project loading when multiple # IOU nodes use the same image. - _loader_cache = {} # image path -> loader command list - _default_values_cache = {} # image path -> (ram, nvram) + _loader_cache: dict[str, list[str]] = {} # image path -> loader command list + _default_values_cache: dict[str, tuple[int, int]] = {} # image path -> (ram, nvram) """ IOU VM implementation. @@ -464,7 +464,11 @@ class IOUVM(BaseNode): ) self._nvram = nvram - @BaseNode.name.setter + @property + def name(self): + return self._name + + @name.setter def name(self, new_name): """ Sets the name of this IOU VM. @@ -949,7 +953,11 @@ class IOUVM(BaseNode): return True return False - @BaseNode.console_type.setter + @property + def console_type(self): + return self._console_type + + @console_type.setter def console_type(self, new_console_type): """ Sets the console type for this IOU VM. diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index 72c6552d5..2a7ec02d8 100644 --- a/gns3server/controller/link.py +++ b/gns3server/controller/link.py @@ -365,7 +365,21 @@ class Link: raise NotImplementedError - async def start_marker(self, name, bpf, tag=None, direction=None, capture_node_id=None, enabled=True): + async def start_marker( + self, + name, + bpf, + tag=None, + direction=None, + data_link_type="DLT_EN10MB", + capture_node_id=None, + color=None, + highlight_duration=None, + enabled=True, + inherited_from=None, + dump=True, + memory_only=False, + ): """ Attach a traffic-insight marker to this link (base — UDPLink overrides). """ @@ -377,7 +391,19 @@ class Link: """ raise NotImplementedError - async def update_marker(self, name, bpf=None, tag=None, enabled=None, direction=_UNSET): + async def update_marker( + self, + name, + bpf=None, + tag=None, + enabled=None, + direction=_UNSET, + color=None, + highlight_duration=None, + inherited=False, + dump=True, + memory_only=False, + ): """ Update an existing marker's BPF, tag, or enabled flag. 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/db/models/acl.py b/gns3server/db/models/acl.py index 32f7f7248..fc75eedac 100644 --- a/gns3server/db/models/acl.py +++ b/gns3server/db/models/acl.py @@ -29,7 +29,7 @@ class ACE(BaseTable): __tablename__ = "acl" ace_id = Column(GUID, primary_key=True, default=generate_uuid) - ace_type: str = Column(String) + ace_type = Column(String) path = Column(String) propagate = Column(Boolean, default=True) allowed = Column(Boolean, default=True) diff --git a/gns3server/db/models/base.py b/gns3server/db/models/base.py index cf726ec79..1802c1771 100644 --- a/gns3server/db/models/base.py +++ b/gns3server/db/models/base.py @@ -21,11 +21,10 @@ from fastapi.encoders import jsonable_encoder from sqlalchemy import Column, DateTime, func, inspect from sqlalchemy.types import TypeDecorator, CHAR, VARCHAR from sqlalchemy.dialects.postgresql import UUID -from sqlalchemy.orm import as_declarative +from sqlalchemy.orm import DeclarativeBase -@as_declarative() -class Base: +class Base(DeclarativeBase): def asdict(self): return {c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs} diff --git a/gns3server/schemas/controller/templates/dynamips_templates.py b/gns3server/schemas/controller/templates/dynamips_templates.py index 90180d99b..69505051d 100644 --- a/gns3server/schemas/controller/templates/dynamips_templates.py +++ b/gns3server/schemas/controller/templates/dynamips_templates.py @@ -30,12 +30,12 @@ from typing import Optional from enum import Enum -class DynamipsTemplate(TemplateBase): +class DynamipsTemplateBase(TemplateBase): category: Optional[Category] = Category.router default_name_format: Optional[str] = "R{0}" symbol: Optional[str] = "router" - platform: DynamipsPlatform = Field(..., description="Cisco router platform") - image: str = Field(..., description="Path to the IOS image") + platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") + image: Optional[str] = Field(None, description="Path to the IOS image") exec_area: Optional[int] = Field(64, ge=0, description="Exec area value") mmap: Optional[bool] = Field(True, description="MMAP feature") mac_addr: Optional[str] = Field( @@ -67,7 +67,12 @@ class DynamipsTemplate(TemplateBase): wic2: Optional[DynamipsWics] = Field(None, description="Network module WIC slot 2") -class C7200DynamipsTemplate(DynamipsTemplate): +class DynamipsTemplate(DynamipsTemplateBase): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") + + +class C7200DynamipsTemplateUpdate(DynamipsTemplateBase): ram: Optional[int] = Field(512, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(512, gt=0, description="Amount of NVRAM in KB") npe: Optional[DynamipsNPE] = Field(DynamipsNPE.npe_400, description="NPE model") @@ -75,33 +80,33 @@ class C7200DynamipsTemplate(DynamipsTemplate): sparsemem: Optional[bool] = Field(True, description="Sparse memory feature") -class C7200DynamipsTemplateUpdate(C7200DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C7200DynamipsTemplate(C7200DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") -class C3725DynamipsTemplate(DynamipsTemplate): +class C3725DynamipsTemplateUpdate(DynamipsTemplateBase): ram: Optional[int] = Field(128, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB") iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage") sparsemem: Optional[bool] = Field(True, description="Sparse memory feature") -class C3725DynamipsTemplateUpdate(C3725DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C3725DynamipsTemplate(C3725DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") -class C3745DynamipsTemplate(DynamipsTemplate): +class C3745DynamipsTemplateUpdate(DynamipsTemplateBase): ram: Optional[int] = Field(256, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB") iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage") sparsemem: Optional[bool] = Field(True, description="Sparse memory feature") -class C3745DynamipsTemplateUpdate(C3745DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C3745DynamipsTemplate(C3745DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") class C3600ChassisType(str, Enum): @@ -110,29 +115,29 @@ class C3600ChassisType(str, Enum): chassis_3660 = "3660" -class C3600DynamipsTemplate(DynamipsTemplate): - chassis: Optional[C3600ChassisType] = Field("c3660", description="Chassis type") +class C3600DynamipsTemplateUpdate(DynamipsTemplateBase): + chassis: Optional[C3600ChassisType] = Field(C3600ChassisType.chassis_3660, description="Chassis type") ram: Optional[int] = Field(192, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB") iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage") sparsemem: Optional[bool] = Field(True, description="Sparse memory feature") -class C3600DynamipsTemplateUpdate(C3600DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C3600DynamipsTemplate(C3600DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") -class C2691DynamipsTemplate(DynamipsTemplate): +class C2691DynamipsTemplateUpdate(DynamipsTemplateBase): ram: Optional[int] = Field(192, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB") iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage") sparsemem: Optional[bool] = Field(True, description="Sparse memory feature") -class C2691DynamipsTemplateUpdate(C2691DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C2691DynamipsTemplate(C2691DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") class C2600ChassisType(str, Enum): @@ -147,17 +152,17 @@ class C2600ChassisType(str, Enum): chassis_2651xm = "2651XM" -class C2600DynamipsTemplate(DynamipsTemplate): - chassis: Optional[C2600ChassisType] = Field("2651XM", description="Chassis type") +class C2600DynamipsTemplateUpdate(DynamipsTemplateBase): + chassis: Optional[C2600ChassisType] = Field(C2600ChassisType.chassis_2651xm, description="Chassis type") ram: Optional[int] = Field(160, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB") iomem: Optional[int] = Field(15, ge=0, le=100, description="I/O memory percentage") sparsemem: Optional[bool] = Field(True, description="Sparse memory feature") -class C2600DynamipsTemplateUpdate(C2600DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C2600DynamipsTemplate(C2600DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") class C1700ChassisType(str, Enum): @@ -168,14 +173,14 @@ class C1700ChassisType(str, Enum): chassis_1760 = "1760" -class C1700DynamipsTemplate(DynamipsTemplate): - chassis: Optional[C1700ChassisType] = Field("1760", description="Chassis type") +class C1700DynamipsTemplateUpdate(DynamipsTemplateBase): + chassis: Optional[C1700ChassisType] = Field(C1700ChassisType.chassis_1760, description="Chassis type") ram: Optional[int] = Field(160, gt=0, description="Amount of RAM in MB") nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB") iomem: Optional[int] = Field(15, ge=0, le=100, description="I/O memory percentage") sparsemem: Optional[bool] = Field(False, description="Sparse memory feature") -class C1700DynamipsTemplateUpdate(C1700DynamipsTemplate): - platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform") - image: Optional[str] = Field(None, description="Path to the IOS image") +class C1700DynamipsTemplate(C1700DynamipsTemplateUpdate): + platform: DynamipsPlatform = Field(..., description="Cisco router platform") + image: str = Field(..., description="Path to the IOS image") diff --git a/gns3server/services/authentication.py b/gns3server/services/authentication.py index 3d165c55d..c507c01a8 100644 --- a/gns3server/services/authentication.py +++ b/gns3server/services/authentication.py @@ -73,21 +73,23 @@ class AuthService: encoded_jwt = jwt.encode({"alg": algorithm}, to_encode, key) return encoded_jwt - def create_access_token(self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0) -> str: + def create_access_token( + self, username, token_version: int = 0, secret_key: Optional[str] = None, expires_in: int = 0 + ) -> str: if not expires_in: expires_in = Config.instance().settings.Controller.jwt_access_token_expire_minutes return self._create_token(username, token_version, "access", expires_in, secret_key) def create_refresh_token( - self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0 + self, username, token_version: int = 0, secret_key: Optional[str] = None, expires_in: int = 0 ) -> str: if not expires_in: expires_in = Config.instance().settings.Controller.jwt_refresh_token_expire_minutes return self._create_token(username, token_version, "refresh", expires_in, secret_key) - def get_token_data(self, token: str, secret_key: str = None) -> TokenData: + def get_token_data(self, token: str, secret_key: Optional[str] = None) -> TokenData: def auth_error(detail: str) -> HTTPException: return HTTPException( @@ -105,7 +107,7 @@ class AuthService: key = OctKey.import_key(secret_key) try: payload = jwt.decode(token, key, algorithms=[algorithm]) - username: str = payload.claims.get("sub") + username: Optional[str] = payload.claims.get("sub") if username is None: raise auth_error("Invalid token: missing subject claim") # Validate the exp claim — joserfc does not validate time-based claims by default @@ -123,5 +125,5 @@ class AuthService: raise auth_error(f"Invalid token ({type(e).__name__})") return token_data - def get_username_from_token(self, token: str, secret_key: str = None) -> Optional[str]: + def get_username_from_token(self, token: str, secret_key: Optional[str] = None) -> Optional[str]: return self.get_token_data(token, secret_key).username diff --git a/gns3server/utils/zipfile_zstd/__init__.py b/gns3server/utils/zipfile_zstd/__init__.py index 4aae04421..f43df1493 100644 --- a/gns3server/utils/zipfile_zstd/__init__.py +++ b/gns3server/utils/zipfile_zstd/__init__.py @@ -2,7 +2,7 @@ from . import _zipfile from zipfile import * -from zipfile import ( +from ._zipfile import ( ZIP_ZSTANDARD, ZSTANDARD_VERSION, ) diff --git a/gns3server/utils/zipfile_zstd/_zipfile.py b/gns3server/utils/zipfile_zstd/_zipfile.py index 2d5d200fa..e4bb61b03 100644 --- a/gns3server/utils/zipfile_zstd/_zipfile.py +++ b/gns3server/utils/zipfile_zstd/_zipfile.py @@ -1,13 +1,18 @@ -import zipfile +import zipfile as _zipfile_module import zstandard as zstd import inspect +from typing import Any from ._patcher import patch +zipfile: Any = _zipfile_module -zipfile.ZIP_ZSTANDARD = 93 +ZIP_ZSTANDARD = 93 +ZSTANDARD_VERSION = 20 + +zipfile.ZIP_ZSTANDARD = ZIP_ZSTANDARD zipfile.compressor_names[zipfile.ZIP_ZSTANDARD] = "zstandard" -zipfile.ZSTANDARD_VERSION = 20 +zipfile.ZSTANDARD_VERSION = ZSTANDARD_VERSION @patch(zipfile, "_check_compression") diff --git a/pyproject.toml b/pyproject.toml index bab14b8c2..6677635c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -295,13 +295,12 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"] # 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.agent_service", # 8 "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 @@ -313,12 +312,8 @@ module = [ "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", # 34 "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 @@ -333,7 +328,6 @@ module = [ "gns3server.api.routes.compute.nat_nodes", # 10 "gns3server.api.routes.compute.notifications", # 4 "gns3server.api.routes.compute.projects", # 7 - "gns3server.api.routes.compute.qemu_nodes", # 18 "gns3server.api.routes.compute.virtualbox_nodes", # 20 "gns3server.api.routes.compute.vmware_nodes", # 10 "gns3server.api.routes.compute.vpcs_nodes", # 9 @@ -347,7 +341,6 @@ module = [ "gns3server.api.routes.controller.drawings", # 1 "gns3server.api.routes.controller.groups", # 6 "gns3server.api.routes.controller.images", # 9 - "gns3server.api.routes.controller.links", # 27 "gns3server.api.routes.controller.llm_model_configs", # 32 "gns3server.api.routes.controller.nodes", # 13 "gns3server.api.routes.controller.pools", # 8 @@ -359,7 +352,6 @@ module = [ "gns3server.api.routes.controller.users", # 14 "gns3server.api.server", # 2 "gns3server.compute.dynamips.nodes.router", # 2 - "gns3server.compute.iou.iou_vm", # 4 "gns3server.compute.iou.utils.iou_export", # 2 "gns3server.compute.iou.utils.iou_import", # 2 "gns3server.compute.virtualbox.virtualbox_vm", # 1 @@ -367,14 +359,6 @@ module = [ "gns3server.controller.appliance", # 2 "gns3server.controller.appliance_manager", # 5 "gns3server.controller.project", # 2 - "gns3server.controller.topology", # 1 - "gns3server.db.models.acl", # 1 - "gns3server.db.models.images", # 1 - "gns3server.db.models.pools", # 1 - "gns3server.db.models.privileges", # 2 - "gns3server.db.models.roles", # 1 - "gns3server.db.models.templates", # 22 - "gns3server.db.models.users", # 3 "gns3server.db.repositories.api_keys", # 8 "gns3server.db.repositories.computes", # 10 "gns3server.db.repositories.images", # 14 @@ -384,7 +368,6 @@ module = [ "gns3server.db.repositories.templates", # 6 "gns3server.db.repositories.users", # 12 "gns3server.db.tasks", # 12 - "gns3server.db_migrations.env", # 1 "gns3server.schemas.compute.atm_switch_nodes", # 5 "gns3server.schemas.compute.cloud_nodes", # 1 "gns3server.schemas.compute.docker_nodes", # 2 @@ -402,12 +385,10 @@ module = [ "gns3server.schemas.controller.projects", # 2 "gns3server.schemas.controller.settings", # 3 "gns3server.schemas.controller.templates.docker_templates", # 1 - "gns3server.schemas.controller.templates.dynamips_templates", # 17 "gns3server.schemas.controller.templates.iou_templates", # 1 "gns3server.schemas.controller.templates.virtualbox_templates", # 1 "gns3server.schemas.controller.templates.vmware_templates", # 1 "gns3server.schemas.qemu_disk_image", # 2 - "gns3server.services.authentication", # 5 "gns3server.services.computes", # 4 "gns3server.services.templates", # 18 "gns3server.utils.asyncio.aiozipstream", # 8 @@ -416,8 +397,6 @@ module = [ "gns3server.utils.images", # 4 "gns3server.utils.packet_filter_validation", # 16 "gns3server.utils.websocket_to_websocket", # 12 - "gns3server.utils.zipfile_zstd", # 2 "gns3server.utils.zipfile_zstd._patcher", # 1 - "gns3server.utils.zipfile_zstd._zipfile", # 5 ] ignore_errors = true diff --git a/tests/agent/mcp/test_tool_params.py b/tests/agent/mcp/test_tool_params.py index 258436462..05c7de063 100644 --- a/tests/agent/mcp/test_tool_params.py +++ b/tests/agent/mcp/test_tool_params.py @@ -193,6 +193,9 @@ def _initial_params_keys(fn_node): and isinstance(stmt.value, ast.Dict) ): return _dict_literal_keys(stmt.value) + if isinstance(stmt, ast.AnnAssign): + if isinstance(stmt.target, ast.Name) and stmt.target.id == "params" and isinstance(stmt.value, ast.Dict): + return _dict_literal_keys(stmt.value) return None