Merge branch '3.1' into mypy-api-routes-compute-3

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