Merge branch '3.1' into mypy-api-routes

gns3server/api/routes/compute/atm_switch_nodes.py

@@ -18,6 +18,7 @@
 API routes for ATM switch nodes.
 """
 
+from typing import Any
 import os
 
 from fastapi import APIRouter, Depends, Body, Path, status, HTTPException
@@ -29,7 +30,9 @@ from gns3server import schemas
 from gns3server.compute.dynamips import Dynamips
 from gns3server.compute.dynamips.nodes.atm_switch import ATMSwitch
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or ATM switch node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or ATM switch node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -57,13 +60,13 @@ async def create_atm_switch(project_id: UUID, node_data: schemas.ATMSwitchCreate
 
     # Use the Dynamips ATM switch to simulate this node
     dynamips_manager = Dynamips.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     node = await dynamips_manager.create_node(
-        node_data.get("name"),
+        data.get("name"),
         str(project_id),
-        node_data.get("node_id"),
+        data.get("node_id"),
         node_type="atm_switch",
-        mappings=node_data.get("mappings"),
+        mappings=data.get("mappings"),
     )
     return node.asdict()
 
@@ -97,11 +100,11 @@ async def update_atm_switch(
     Update an ATM switch node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    if "name" in node_data and node.name != node_data["name"]:
-        await node.set_name(node_data["name"])
-    if "mappings" in node_data:
-        node.mappings = node_data["mappings"]
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    if "name" in data and node.name != data["name"]:
+        await node.set_name(data["name"])
+    if "mappings" in data:
+        node.mappings = data["mappings"]
     node.updated()
     return node.asdict()
 

gns3server/api/routes/compute/capabilities.py

@@ -39,7 +39,8 @@ def get_capabilities(request: Request) -> dict:
         node_types.extend(module.node_types())
 
     # record the controller hostname or IP address
-    request.app.state.controller_host = request.client.host
+    if request.client:
+        request.app.state.controller_host = request.client.host
 
     return {
         "version": __version__,

gns3server/api/routes/compute/cloud_nodes.py

@@ -23,14 +23,16 @@ import os
 from fastapi import APIRouter, Depends, Path, status, HTTPException
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
-from typing import Union
+from typing import Any, Union
 from uuid import UUID
 
 from gns3server import schemas
 from gns3server.compute.builtin import Builtin
 from gns3server.compute.builtin.nodes.cloud import Cloud
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or cloud node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or cloud node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -57,21 +59,21 @@ async def create_cloud(project_id: UUID, node_data: schemas.CloudCreate) -> sche
     """
 
     builtin_manager = Builtin.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     node = await builtin_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
+        data.get("node_id"),
         node_type="cloud",
-        ports=node_data.get("ports_mapping"),
+        ports=data.get("ports_mapping"),
     )
 
     # add the remote console settings
-    node.remote_console_host = node_data.get("remote_console_host", node.remote_console_host)
-    node.remote_console_port = node_data.get("remote_console_port", node.remote_console_port)
-    node.remote_console_type = node_data.get("remote_console_type", node.remote_console_type)
-    node.remote_console_http_path = node_data.get("remote_console_http_path", node.remote_console_http_path)
-    node.usage = node_data.get("usage", "")
+    node.remote_console_host = data.get("remote_console_host", node.remote_console_host)
+    node.remote_console_port = data.get("remote_console_port", node.remote_console_port)
+    node.remote_console_type = data.get("remote_console_type", node.remote_console_type)
+    node.remote_console_http_path = data.get("remote_console_http_path", node.remote_console_http_path)
+    node.usage = data.get("usage", "")
     return node.asdict()
 
 
@@ -90,8 +92,8 @@ async def update_cloud(node_data: schemas.CloudUpdate, node: Cloud = Depends(dep
     Update a cloud node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    for name, value in node_data.items():
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    for name, value in data.items():
         if hasattr(node, name) and getattr(node, name) != value:
             setattr(node, name, value)
     node.updated()
@@ -178,7 +180,7 @@ async def update_cloud_nio(
 
     nio = node.get_nio(port_number)
     nio.filters.clear()
-    if nio_data.filters:
+    if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters:
         nio.filters = nio_data.filters
     # NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers.
     nio.markers = getattr(nio_data, "markers", None) or {}

gns3server/api/routes/compute/compute.py

@@ -163,7 +163,7 @@ def compute_statistics() -> dict:
 
 @router.get("/qemu/capabilities")
 async def get_qemu_capabilities() -> dict:
-    capabilities = {"kvm": []}
+    capabilities: dict[str, list] = {"kvm": []}
     kvms = await Qemu.get_kvm_archs()
     if kvms:
         capabilities["kvm"] = kvms

gns3server/api/routes/compute/dependencies/authentication.py

@@ -91,9 +91,9 @@ async def ws_compute_authentication(websocket: WebSocket) -> Union[None, WebSock
         if not separator:
             raise invalid_user_credentials_exc
 
-        username = secrets.compare_digest(username, server_settings.compute_username)
-        password = secrets.compare_digest(password, server_settings.compute_password.get_secret_value())
-        if not (username and password):
+        username_valid = secrets.compare_digest(username, server_settings.compute_username)
+        password_valid = secrets.compare_digest(password, server_settings.compute_password.get_secret_value())
+        if not (username_valid and password_valid):
             raise invalid_user_credentials_exc
 
     except HTTPException as e:
@@ -101,5 +101,6 @@ async def ws_compute_authentication(websocket: WebSocket) -> Union[None, WebSock
         websocket_error = {"action": "log.error", "event": {"message": err_msg}}
         await websocket.send_json(websocket_error)
         log.error(err_msg)
-        return await websocket.close(code=1008)
+        await websocket.close(code=1008)
+        return None
     return websocket

gns3server/api/routes/compute/docker_nodes.py

@@ -24,14 +24,16 @@ from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
 from uuid import UUID
-from typing import Union
+from typing import Any, Union
 
 from gns3server import schemas
 from gns3server.compute.docker import Docker
 from gns3server.compute.docker.docker_vm import DockerVM
 from .dependencies.authentication import compute_authentication, ws_compute_authentication
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Docker node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or Docker node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -53,13 +55,13 @@ def dep_node(project_id: UUID, node_id: UUID) -> DockerVM:
     responses={409: {"model": schemas.ErrorMessage, "description": "Could not create Docker node"}},
     dependencies=[Depends(compute_authentication)],
 )
-async def create_docker_node(project_id: UUID, node_data: schemas.DockerCreate) -> schemas.Docker:
+async def create_docker_node(project_id: UUID, node_create: schemas.DockerCreate) -> schemas.Docker:
     """
     Create a new Docker node.
     """
 
     docker_manager = Docker.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    node_data = jsonable_encoder(node_create, exclude_unset=True)
     container = await docker_manager.create_node(
         node_data.pop("name"),
         str(project_id),
@@ -124,7 +126,7 @@ def get_docker_node(node: DockerVM = Depends(dep_node)) -> schemas.Docker:
 
 
 @router.put("/{node_id}", response_model=schemas.Docker, dependencies=[Depends(compute_authentication)])
-async def update_docker_node(node_data: schemas.DockerUpdate, node: DockerVM = Depends(dep_node)) -> schemas.Docker:
+async def update_docker_node(node_update: schemas.DockerUpdate, node: DockerVM = Depends(dep_node)) -> schemas.Docker:
     """
     Update a Docker node.
     """
@@ -152,7 +154,7 @@ async def update_docker_node(node_data: schemas.DockerUpdate, node: DockerVM = D
     ]
 
     changed = False
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    node_data = jsonable_encoder(node_update, exclude_unset=True)
     for prop in props:
         # hasattr: startup_config_content only exists on IOLDockerVM
         if prop in node_data and hasattr(node, prop) and node_data[prop] != getattr(node, prop):

gns3server/api/routes/compute/dynamips_nodes.py

@@ -23,7 +23,7 @@ import os
 from fastapi import APIRouter, WebSocket, Body, Depends, status, HTTPException
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
-from typing import List, Union
+from typing import Any, List, Optional, Union
 from uuid import UUID
 
 from gns3server.compute.dynamips import Dynamips
@@ -32,7 +32,9 @@ from gns3server import schemas
 
 from .dependencies.authentication import compute_authentication, ws_compute_authentication
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Dynamips node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or Dynamips node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -64,25 +66,26 @@ async def create_router(project_id: UUID, node_data: schemas.DynamipsCreate) ->
 
     dynamips_manager = Dynamips.instance()
     platform = node_data.platform
+    chassis: Optional[str]
     if not node_data.chassis and platform in DEFAULT_CHASSIS:
         chassis = DEFAULT_CHASSIS[platform]
     else:
         chassis = node_data.chassis
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     vm = await dynamips_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
-        dynamips_id=node_data.get("dynamips_id"),
+        data.get("node_id"),
+        dynamips_id=data.get("dynamips_id"),
         platform=platform,
-        console=node_data.get("console"),
-        console_type=node_data.get("console_type", "telnet"),
-        aux=node_data.get("aux"),
-        aux_type=node_data.pop("aux_type", "none"),
+        console=data.get("console"),
+        console_type=data.get("console_type", "telnet"),
+        aux=data.get("aux"),
+        aux_type=data.pop("aux_type", "none"),
         chassis=chassis,
         node_type="dynamips",
     )
-    await dynamips_manager.update_vm_settings(vm, node_data)
+    await dynamips_manager.update_vm_settings(vm, data)
     return vm.asdict()
 
 

gns3server/api/routes/compute/ethernet_hub_nodes.py

@@ -18,6 +18,7 @@
 API routes for Ethernet hub nodes.
 """
 
+from typing import Any
 import os
 
 from fastapi import APIRouter, Depends, Body, Path, status, HTTPException
@@ -29,7 +30,9 @@ from gns3server.compute.dynamips import Dynamips
 from gns3server.compute.dynamips.nodes.ethernet_hub import EthernetHub
 from gns3server import schemas
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet hub node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet hub node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -57,13 +60,13 @@ async def create_ethernet_hub(project_id: UUID, node_data: schemas.EthernetHubCr
 
     # Use the Dynamips Ethernet hub to simulate this node
     dynamips_manager = Dynamips.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     node = await dynamips_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
+        data.get("node_id"),
         node_type="ethernet_hub",
-        ports=node_data.get("ports_mapping"),
+        ports=data.get("ports_mapping"),
     )
     return node.asdict()
 
@@ -97,11 +100,11 @@ async def update_ethernet_hub(
     Update an Ethernet hub.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    if "name" in node_data and node.name != node_data["name"]:
-        await node.set_name(node_data["name"])
-    if "ports_mapping" in node_data:
-        node.ports_mapping = node_data["ports_mapping"]
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    if "name" in data and node.name != data["name"]:
+        await node.set_name(data["name"])
+    if "ports_mapping" in data:
+        node.ports_mapping = data["ports_mapping"]
     node.updated()
     return node.asdict()
 

gns3server/api/routes/compute/ethernet_switch_nodes.py

@@ -22,6 +22,7 @@ through uBridge's ``brctl`` module (see
 ``gns3server.compute.builtin.nodes.ethernet_switch``).
 """
 
+from typing import Any
 import os
 
 from fastapi import APIRouter, Depends, Body, Path, status, HTTPException
@@ -33,7 +34,9 @@ from gns3server.compute.builtin import Builtin
 from gns3server.compute.builtin.nodes.ethernet_switch import EthernetSwitch
 from gns3server import schemas
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet switch node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or Ethernet switch node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -60,17 +63,17 @@ async def create_ethernet_switch(project_id: UUID, node_data: schemas.EthernetSw
     """
 
     builtin_manager = Builtin.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     node = await builtin_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
-        console=node_data.get("console"),
-        console_type=node_data.get("console_type"),
+        data.get("node_id"),
+        console=data.get("console"),
+        console_type=data.get("console_type"),
         node_type="ethernet_switch",
-        ports=node_data.get("ports_mapping"),
+        ports=data.get("ports_mapping"),
     )
-    node.usage = node_data.get("usage", "")
+    node.usage = data.get("usage", "")
     return node.asdict()
 
 
@@ -100,16 +103,16 @@ async def update_ethernet_switch(
     Update an Ethernet switch.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    if "name" in node_data and node.name != node_data["name"]:
-        node.name = node_data["name"]
-    if "usage" in node_data:
-        node.usage = node_data["usage"]
-    if "ports_mapping" in node_data:
-        node.ports_mapping = node_data["ports_mapping"]
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    if "name" in data and node.name != data["name"]:
+        node.name = data["name"]
+    if "usage" in data:
+        node.usage = data["usage"]
+    if "ports_mapping" in data:
+        node.ports_mapping = data["ports_mapping"]
         await node.update_port_settings()
-    if "console_type" in node_data:
-        node.console_type = node_data["console_type"]
+    if "console_type" in data:
+        node.console_type = data["console_type"]
     node.updated()
     return node.asdict()
 

gns3server/api/routes/compute/frame_relay_switch_nodes.py

@@ -18,6 +18,7 @@
 API routes for Frame Relay switch nodes.
 """
 
+from typing import Any
 import os
 
 from fastapi import APIRouter, Depends, Body, Path, status, HTTPException
@@ -29,7 +30,9 @@ from gns3server import schemas
 from gns3server.compute.dynamips import Dynamips
 from gns3server.compute.dynamips.nodes.frame_relay_switch import FrameRelaySwitch
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Frame Relay switch node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or Frame Relay switch node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -59,13 +62,13 @@ async def create_frame_relay_switch(
 
     # Use the Dynamips Frame Relay switch to simulate this node
     dynamips_manager = Dynamips.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     node = await dynamips_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
+        data.get("node_id"),
         node_type="frame_relay_switch",
-        mappings=node_data.get("mappings"),
+        mappings=data.get("mappings"),
     )
     return node.asdict()
 
@@ -99,11 +102,11 @@ async def update_frame_relay_switch(
     Update an Frame Relay switch node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    if "name" in node_data and node.name != node_data["name"]:
-        await node.set_name(node_data["name"])
-    if "mappings" in node_data:
-        node.mappings = node_data["mappings"]
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    if "name" in data and node.name != data["name"]:
+        await node.set_name(data["name"])
+    if "mappings" in data:
+        node.mappings = data["mappings"]
     node.updated()
     return node.asdict()
 

gns3server/api/routes/compute/iou_nodes.py

@@ -23,7 +23,7 @@ import os
 from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
-from typing import Union
+from typing import Any, Union
 from uuid import UUID
 
 from gns3server import schemas
@@ -32,7 +32,9 @@ from gns3server.compute.iou.iou_vm import IOUVM
 
 from .dependencies.authentication import compute_authentication, ws_compute_authentication
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -60,18 +62,18 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch
     """
 
     iou = IOU.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     vm = await iou.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
-        application_id=node_data.get("application_id"),
-        path=node_data.get("path"),
-        console=node_data.get("console"),
-        console_type=node_data.get("console_type", "telnet"),
+        data.get("node_id"),
+        application_id=data.get("application_id"),
+        path=data.get("path"),
+        console=data.get("console"),
+        console_type=data.get("console_type", "telnet"),
     )
 
-    for name, value in node_data.items():
+    for name, value in data.items():
         if hasattr(vm, name) and getattr(vm, name) != value:
             if name == "application_id":
                 continue  # we must ignore this to avoid overwriting the application_id allocated by the controller
@@ -79,7 +81,7 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch
                 continue
             if name == "private_config_content" and (vm.private_config_content and len(vm.private_config_content) > 0):
                 continue
-            if node_data.get("use_default_iou_values") and (name == "ram" or name == "nvram"):
+            if data.get("use_default_iou_values") and (name == "ram" or name == "nvram"):
                 continue
             setattr(vm, name, value)
     return vm.asdict()
@@ -100,8 +102,8 @@ async def update_iou_node(node_data: schemas.IOUUpdate, node: IOUVM = Depends(de
     Update an IOU node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    for name, value in node_data.items():
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    for name, value in data.items():
         if hasattr(node, name) and getattr(node, name) != value:
             if name == "application_id":
                 continue  # we must ignore this to avoid overwriting the application_id allocated by the IOU manager
@@ -147,8 +149,8 @@ async def start_iou_node(start_data: schemas.IOUStart, node: IOUVM = Depends(dep
     Start an IOU node.
     """
 
-    start_data = jsonable_encoder(start_data, exclude_unset=True)
-    for name, value in start_data.items():
+    start_values = jsonable_encoder(start_data, exclude_unset=True)
+    for name, value in start_values.items():
         if hasattr(node, name) and getattr(node, name) != value:
             setattr(node, name, value)
 
@@ -223,7 +225,7 @@ async def update_iou_node_nio(
 
     nio = node.get_nio(adapter_number, port_number)
     nio.filters.clear()
-    if nio_data.filters:
+    if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters:
         nio.filters = nio_data.filters
     # NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers.
     nio.markers = getattr(nio_data, "markers", None) or {}

gns3server/api/routes/compute/nat_nodes.py

@@ -23,14 +23,16 @@ import os
 from fastapi import APIRouter, Depends, Path, status, HTTPException
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
-from typing import Union
+from typing import Any, Union
 from uuid import UUID
 
 from gns3server import schemas
 from gns3server.compute.builtin import Builtin
 from gns3server.compute.builtin.nodes.nat import Nat
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or NAT node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or NAT node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -57,16 +59,16 @@ async def create_nat_node(project_id: UUID, node_data: schemas.NATCreate) -> sch
     """
 
     builtin_manager = Builtin.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     node = await builtin_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
+        data.get("node_id"),
         node_type="nat",
-        ports=node_data.get("ports_mapping"),
+        ports=data.get("ports_mapping"),
     )
 
-    node.usage = node_data.get("usage", "")
+    node.usage = data.get("usage", "")
     return node.asdict()
 
 
@@ -85,8 +87,8 @@ async def update_nat_node(node_data: schemas.NATUpdate, node: Nat = Depends(dep_
     Update a NAT node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    for name, value in node_data.items():
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    for name, value in data.items():
         if hasattr(node, name) and getattr(node, name) != value:
             setattr(node, name, value)
     node.updated()
@@ -170,7 +172,7 @@ async def update_nat_node_nio(
 
     nio = node.get_nio(port_number)
     nio.filters.clear()
-    if nio_data.filters:
+    if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters:
         nio.filters = nio_data.filters
     await node.update_nio(port_number, nio)
     return nio.asdict()

gns3server/api/routes/compute/notifications.py

@@ -39,14 +39,15 @@ async def project_ws_notifications(websocket: Union[None, WebSocket] = Depends(w
     """
 
     if websocket:
-        log.info(f"New client {websocket.client.host}:{websocket.client.port} has connected to compute WebSocket")
+        client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown"
+        log.info(f"New client {client} has connected to compute WebSocket")
         try:
             with NotificationManager.instance().queue() as queue:
                 while True:
                     notification = await queue.get_json(5)
                     await websocket.send_text(notification)
         except (ConnectionClosed, WebSocketDisconnect):
-            log.info(f"Client {websocket.client.host}:{websocket.client.port} has disconnected from compute WebSocket")
+            log.info(f"Client {client} has disconnected from compute WebSocket")
         except WebSocketException as e:
             log.warning(f"Error while sending to controller event to WebSocket client: {e}")
 

gns3server/api/routes/compute/projects.py

@@ -44,7 +44,7 @@ from gns3server import schemas
 router = APIRouter()
 
 # How many clients have subscribed to notifications
-_notifications_listening = {}
+_notifications_listening: dict[str, int] = {}
 
 
 def dep_project(project_id: UUID) -> Project:
@@ -74,12 +74,12 @@ def create_compute_project(project_data: schemas.ProjectCreate) -> schemas.Proje
     """
 
     pm = ProjectManager.instance()
-    project_data = jsonable_encoder(project_data, exclude_unset=True)
+    data = jsonable_encoder(project_data, exclude_unset=True)
     project = pm.create_project(
-        name=project_data.get("name"),
-        path=project_data.get("path"),
-        project_id=project_data.get("project_id"),
-        variables=project_data.get("variables", None),
+        name=data.get("name"),
+        path=data.get("path"),
+        project_id=data.get("project_id"),
+        variables=data.get("variables", None),
     )
     return project.asdict()
 
@@ -261,7 +261,7 @@ async def create_batch_nios(
     # This is what makes builtin L2 nodes (ethernet_switch/hub/cloud/nat)
     # start their uBridge concurrently during project open instead of one
     # at a time (~0.5s each for fork + socket connect).
-    per_node = {}
+    per_node: dict[str, list[schemas.BatchNIOEntry]] = {}
     for entry in batch.nios:
         per_node.setdefault(entry.node_id, []).append(entry)
 
@@ -308,7 +308,7 @@ async def update_batch_nios(
     # Group entries by node so that different nodes' uBridge processes are
     # updated in parallel (each node has its own AF_UNIX socket). Within a
     # node entries are serial to respect the per-node uBridge command lock.
-    per_node = {}
+    per_node: dict[str, list[schemas.BatchNIOEntry]] = {}
     for entry in batch.nios:
         per_node.setdefault(entry.node_id, []).append(entry)
 

gns3server/api/routes/compute/virtualbox_nodes.py

@@ -24,7 +24,7 @@ from fastapi import APIRouter, WebSocket, Depends, Path, status
 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.virtualbox import VirtualBox
@@ -33,7 +33,9 @@ from gns3server.compute.virtualbox.virtualbox_vm import VirtualBoxVM
 
 from .dependencies.authentication import compute_authentication, ws_compute_authentication
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or VirtualBox node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or VirtualBox node"}
+}
 
 router = APIRouter(responses=responses, deprecated=True)
 
@@ -61,24 +63,24 @@ async def create_virtualbox_node(project_id: UUID, node_data: schemas.VirtualBox
     """
 
     vbox_manager = VirtualBox.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     vm = await vbox_manager.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.get("node_id"),
-        node_data.pop("vmname"),
-        linked_clone=node_data.pop("linked_clone", False),
-        console=node_data.get("console", None),
-        console_type=node_data.get("console_type", "telnet"),
-        adapters=node_data.get("adapters", 0),
+        data.get("node_id"),
+        data.pop("vmname"),
+        linked_clone=data.pop("linked_clone", False),
+        console=data.get("console", None),
+        console_type=data.get("console_type", "telnet"),
+        adapters=data.get("adapters", 0),
     )
 
-    if "ram" in node_data:
-        ram = node_data.pop("ram")
+    if "ram" in data:
+        ram = data.pop("ram")
         if ram != vm.ram:
             await vm.set_ram(ram)
 
-    for name, value in node_data.items():
+    for name, value in data.items():
         if name != "node_id":
             if hasattr(vm, name) and getattr(vm, name) != value:
                 setattr(vm, name, value)
@@ -103,10 +105,10 @@ async def update_virtualbox_node(
     Update a VirtualBox node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
-    if "name" in node_data:
-        name = node_data.pop("name")
-        vmname = node_data.pop("vmname", None)
+    data = jsonable_encoder(node_data, exclude_unset=True)
+    if "name" in data:
+        name = data.pop("name")
+        vmname = data.pop("vmname", None)
         if name != node.name:
             oldname = node.name
             node.name = name
@@ -118,20 +120,20 @@ async def update_virtualbox_node(
                     node.updated()
                     raise e
 
-    if "adapters" in node_data:
-        adapters = node_data.pop("adapters")
+    if "adapters" in data:
+        adapters = data.pop("adapters")
         if adapters != node.adapters:
             await node.set_adapters(adapters)
 
-    if "ram" in node_data:
-        ram = node_data.pop("ram")
+    if "ram" in data:
+        ram = data.pop("ram")
         if ram != node.ram:
             await node.set_ram(ram)
 
     # update the console first to avoid issue if updating console type
-    node.console = node_data.pop("console", node.console)
+    node.console = data.pop("console", node.console)
 
-    for name, value in node_data.items():
+    for name, value in data.items():
         if hasattr(node, name) and getattr(node, name) != value:
             setattr(node, name, value)
 

gns3server/api/routes/controller/api_keys.py

@@ -39,7 +39,7 @@ API_KEY_PREFIX = "gns3_"
 API_KEY_BYTES = 32
 
 
-def _generate_api_key(api_key_id: UUID = None) -> tuple[str, str, str, UUID]:
+def _generate_api_key(api_key_id: UUID | None = None) -> tuple[str, str, str, UUID]:
     if api_key_id is None:
         api_key_id = uuid4()
     random_bytes = secrets.token_hex(API_KEY_BYTES)

gns3server/api/routes/controller/chat.py

@@ -23,7 +23,7 @@ Nested under projects: /v3/projects/{project_id}/chat/...
 import json
 import logging
 import uuid
-from typing import List, Optional
+from typing import Any, List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from fastapi.responses import StreamingResponse
@@ -40,7 +40,7 @@ from .dependencies.authentication import get_current_active_user
 
 log = logging.getLogger(__name__)
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Resource not found"}}
+responses: dict[int | str, dict[str, Any]] = {404: {"model": schemas.ErrorMessage, "description": "Resource not found"}}
 
 router = APIRouter(responses=responses)
 

gns3server/api/routes/controller/drawings.py

@@ -20,7 +20,7 @@ API routes for drawings.
 
 from fastapi import APIRouter, Depends, status
 from fastapi.encoders import jsonable_encoder
-from typing import List
+from typing import Any, List
 from uuid import UUID
 
 from gns3server.controller import Controller
@@ -30,7 +30,9 @@ from gns3server import schemas
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Project or drawing not found"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Project or drawing not found"}
+}
 
 router = APIRouter(responses=responses)
 

gns3server/api/routes/controller/nodes.py

@@ -27,7 +27,7 @@ from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, Request,
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
 from fastapi.routing import APIRoute
-from typing import List, Callable, Optional
+from typing import Any, List, Callable, Optional
 from uuid import UUID
 
 from gns3server.controller import Controller
@@ -47,7 +47,7 @@ import logging
 
 log = logging.getLogger(__name__)
 
-node_locks = {}
+node_locks: dict[str, dict[str, Any]] = {}
 
 
 class NodeConcurrency(APIRoute):
@@ -84,7 +84,9 @@ class NodeConcurrency(APIRoute):
         return custom_route_handler
 
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or node"}
+}
 
 router = APIRouter(route_class=NodeConcurrency, responses=responses)
 
@@ -127,7 +129,7 @@ def _check_node_type(node: Node, *required_types: str) -> None:
     },
     dependencies=[Depends(has_privilege("Node.Allocate"))],
 )
-async def create_node(node_data: schemas.NodeCreate, project: Project = Depends(dep_project)) -> schemas.Node:
+async def create_node(node_create: schemas.NodeCreate, project: Project = Depends(dep_project)) -> schemas.Node:
     """
     Create a new node.
 
@@ -135,8 +137,8 @@ async def create_node(node_data: schemas.NodeCreate, project: Project = Depends(
     """
 
     controller = Controller.instance()
-    compute = controller.get_compute(str(node_data.compute_id))
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    compute = controller.get_compute(str(node_create.compute_id))
+    node_data = jsonable_encoder(node_create, exclude_unset=True)
     node = await project.add_node(compute, node_data.pop("name"), node_data.pop("node_id", None), **node_data)
     return node.asdict()
 
@@ -274,14 +276,14 @@ async def get_node(node: Node = Depends(dep_node)) -> schemas.Node:
     response_model_exclude_unset=True,
     dependencies=[Depends(has_privilege("Node.Modify"))],
 )
-async def update_node(node_data: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node:
+async def update_node(node_update: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node:
     """
     Update a node.
 
     Required privilege: Node.Modify
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    node_data = jsonable_encoder(node_update, exclude_unset=True)
 
     # Ignore these because we only use them when creating a node
     node_data.pop("node_id", None)
@@ -654,9 +656,8 @@ async def ws_console(
         return
 
     compute = node.compute
-    log.info(
-        f"New client {websocket.client.host}:{websocket.client.port} has connected to controller console WebSocket"
-    )
+    client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown"
+    log.info(f"New client {client} has connected to controller console WebSocket")
 
     compute_host = compute.host
     try:
@@ -690,9 +691,7 @@ async def ws_console(
                     await ws_console_compute.send_bytes(msg["bytes"])
         except WebSocketDisconnect:
             pass
-        log.info(
-            f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller console WebSocket"
-        )
+        log.info(f"Client {client} has disconnected from controller console WebSocket")
 
     async def ws_send(ws_console_compute):
         """
@@ -709,10 +708,7 @@ async def ws_console(
                     break
         except WebSocketDisconnect:
             # the client disconnected while the compute was still streaming console output
-            log.info(
-                f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller"
-                f" console WebSocket"
-            )
+            log.info(f"Client {client} has disconnected from controller console WebSocket")
 
     try:
         # forward WebSocket data in both directions between the client and the compute console WebSocket
@@ -763,9 +759,8 @@ async def vnc_console(
         return
 
     compute = node.compute
-    log.info(
-        f"New client {websocket.client.host}:{websocket.client.port} has connected to controller VNC console WebSocket"
-    )
+    client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown"
+    log.info(f"New client {client} has connected to controller VNC console WebSocket")
 
     compute_host = compute.host
     try:
@@ -796,10 +791,7 @@ async def vnc_console(
                     await vnc_console_compute.send_bytes(data)
         except WebSocketDisconnect:
             pass
-        log.info(
-            f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller"
-            f" VNC console WebSocket"
-        )
+        log.info(f"Client {client} has disconnected from controller VNC console WebSocket")
 
     async def vnc_send(vnc_console_compute):
         """
@@ -814,10 +806,7 @@ async def vnc_console(
                     break
         except WebSocketDisconnect:
             # the client disconnected while the compute was still streaming VNC console output
-            log.info(
-                f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller"
-                f" VNC console WebSocket"
-            )
+            log.info(f"Client {client} has disconnected from controller VNC console WebSocket")
 
     try:
         # forward WebSocket data in both directions between the client and the compute VNC console WebSocket

gns3server/api/routes/controller/privileges.py

@@ -16,6 +16,7 @@ API route for privileges
 """
 
 from typing import List
+import gns3server.db.models as models
 from gns3server.db.repositories.rbac import RbacRepository
 from .dependencies.database import get_repository
 from fastapi import APIRouter, Depends
@@ -33,7 +34,7 @@ router = APIRouter()
 )
 async def get_privileges(
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
-) -> List[schemas.Privilege]:
+) -> List[models.Privilege]:
     """
     Get all privileges.
 

gns3server/api/routes/controller/roles.py

@@ -21,7 +21,7 @@ API routes for roles.
 
 from fastapi import APIRouter, Depends, status
 from uuid import UUID
-from typing import List
+from typing import List, Optional
 
 from gns3server import schemas
 from gns3server.controller.controller_error import (
@@ -31,6 +31,7 @@ from gns3server.controller.controller_error import (
     ControllerForbiddenError,
 )
 
+import gns3server.db.models as models
 from gns3server.db.repositories.rbac import RbacRepository
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege
@@ -43,7 +44,7 @@ router = APIRouter()
 
 
 @router.get("", response_model=List[schemas.Role], dependencies=[Depends(has_privilege("Role.Audit"))])
-async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[schemas.Role]:
+async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[models.Role]:
     """
     Get all roles.
 
@@ -61,7 +62,7 @@ async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepos
 )
 async def create_role(
     role_create: schemas.RoleCreate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
-) -> schemas.Role:
+) -> Optional[models.Role]:
     """
     Create a new role.
 
@@ -78,7 +79,7 @@ async def create_role(
 async def get_role(
     role_id: UUID,
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
-) -> schemas.Role:
+) -> models.Role:
     """
     Get a role.
 
@@ -94,7 +95,7 @@ async def get_role(
 @router.put("/{role_id}", response_model=schemas.Role, dependencies=[Depends(has_privilege("Role.Modify"))])
 async def update_role(
     role_id: UUID, role_update: schemas.RoleUpdate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
-) -> schemas.Role:
+) -> Optional[models.Role]:
     """
     Update a role.
 
@@ -142,7 +143,7 @@ async def delete_role(
 )
 async def get_role_privileges(
     role_id: UUID, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
-) -> List[schemas.Privilege]:
+) -> List[models.Privilege]:
     """
     Get all role privileges.
 

gns3server/api/routes/controller/snapshots.py

@@ -24,7 +24,7 @@ import logging
 log = logging.getLogger()
 
 from fastapi import APIRouter, Depends, status
-from typing import List
+from typing import Any, List
 from uuid import UUID
 
 from gns3server.controller.project import Project
@@ -35,7 +35,9 @@ from gns3server.controller import Controller
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or snapshot"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or snapshot"}
+}
 
 router = APIRouter(responses=responses)