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-gns3_copilot-plus

Browse Source
This commit is contained in:
Jeremy Grossmann 2026-09-29 13:20:27 +02:00 committed by GitHub
commit 726da8383a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
46 changed files with 185 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gns3server/agent/__init__.py
View File

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

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

8
gns3server/agent/gns3_copilot/chat_sessions_repository.py
View File

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

20
gns3server/agent/gns3_copilot/gns3_client/api_handlers.py
View File

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

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

2
gns3server/agent/gns3_copilot/skills/loader.py
View File

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

18
gns3server/agent/gns3_copilot/skills/manager.py
View File

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

4
gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py
View File

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

4
gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py
View File

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

2
gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py
View File

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

2
gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py
View File

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

2
gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py
View File

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

6
gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py
View File

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

6
gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py
View File

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

6
gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py
View File

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

3
gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py
View File

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

8
gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py
View File

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

5
gns3server/agent/gns3_copilot/utils/message_converters.py
View File

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

4
gns3server/agent/gns3_copilot/utils/parse_tool_content.py
View File

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

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

3
gns3server/api/routes/compute/capabilities.py
View File

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

2
gns3server/api/routes/compute/compute.py
View File

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

9
gns3server/api/routes/compute/dependencies/authentication.py
View File

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

14
gns3server/api/routes/compute/docker_nodes.py
View File

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

25
gns3server/api/routes/compute/dynamips_nodes.py
View File

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

23
gns3server/api/routes/compute/ethernet_hub_nodes.py
View File

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

5
gns3server/api/routes/compute/notifications.py
View File

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

2
gns3server/api/routes/controller/api_keys.py
View File

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

4
gns3server/api/routes/controller/chat.py
View File

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

6
gns3server/api/routes/controller/drawings.py
View File

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

47
gns3server/api/routes/controller/nodes.py
View File

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

3
gns3server/api/routes/controller/privileges.py
View File

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

13
gns3server/api/routes/controller/roles.py
View File

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

6
gns3server/api/routes/controller/snapshots.py
View File

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

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

4
gns3server/schemas/compute/atm_switch_nodes.py
View File

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

4
gns3server/schemas/compute/dynamips_nodes.py
View File

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

7
gns3server/schemas/compute/nat_nodes.py
View File

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

4
gns3server/utils/encryption.py
View File

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

10
gns3server/utils/images.py
View File

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

10
gns3server/utils/packet_filter_validation.py
View File

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

43
pyproject.toml
View File

@ -294,60 +294,24 @@ 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.message_converters", # 2
"gns3server.agent.gns3_copilot.utils.parse_tool_content", # 1
"gns3server.agent.web_wireshark.manage_wireshark", # 1
"gns3server.agent.web_wireshark.manager", # 9
"gns3server.agent.web_wireshark.stats", # 4
"gns3server.api.routes.compute.capabilities", # 1
"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.projects", # 7
"gns3server.api.routes.compute.virtualbox_nodes", # 20
"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
@ -359,7 +323,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
@ -369,13 +332,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
@ -393,10 +353,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
]