NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-09-30 07:40:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch '3.1' into mypy-agent-copilot-skills

Browse Source
This commit is contained in:
Jeremy Grossmann 2026-09-29 12:30:40 +02:00 committed by GitHub
commit 7b40064658
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
22 changed files with 199 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gns3server/agent/gns3_copilot/agent/context_manager.py
View File

@ -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,

4
gns3server/agent/gns3_copilot/agent/model_factory.py
View File

@ -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"],

37
gns3server/agent/gns3_copilot/agent_service.py
View File

@ -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

2
gns3server/agent/gns3_copilot/gns3_client/connector_factory.py
View File

@ -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

6
gns3server/agent/gns3_copilot/gns3_client/project_inventory.py
View File

@ -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

54
gns3server/agent/mcp/__init__.py
View File

@ -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:

5
gns3server/agent/web_wireshark/manage_wireshark.py
View File

@ -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__)

16
gns3server/agent/web_wireshark/manager.py
View File

@ -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,
):

4
gns3server/agent/web_wireshark/stats.py
View File

@ -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()

44
gns3server/api/routes/compute/qemu_nodes.py
View File

@ -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()

18
gns3server/api/routes/controller/links.py
View File

@ -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.

16
gns3server/compute/iou/iou_vm.py
View File

@ -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.

30
gns3server/controller/link.py
View File

@ -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.

2
gns3server/controller/topology.py
View File

@ -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):

2
gns3server/db/models/acl.py
View File

@ -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)

5
gns3server/db/models/base.py
View File

@ -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}

73
gns3server/schemas/controller/templates/dynamips_templates.py
View File

@ -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")

12
gns3server/services/authentication.py
View File

@ -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

2
gns3server/utils/zipfile_zstd/__init__.py
View File

@ -2,7 +2,7 @@
from . import _zipfile
from zipfile import *
from zipfile import (
from ._zipfile import (
ZIP_ZSTANDARD,
ZSTANDARD_VERSION,
)

11
gns3server/utils/zipfile_zstd/_zipfile.py
View File

@ -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")

25
pyproject.toml
View File

@ -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

3
tests/agent/mcp/test_tool_params.py
View File

@ -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