Merge branch '3.1' into image-manager-out-of-sync

gns3server/services/authentication.py

@@ -73,21 +73,23 @@ class AuthService:
         encoded_jwt = jwt.encode({"alg": algorithm}, to_encode, key)
         return encoded_jwt
 
-    def create_access_token(self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0) -> str:
+    def create_access_token(
+        self, username, token_version: int = 0, secret_key: Optional[str] = None, expires_in: int = 0
+    ) -> str:
 
         if not expires_in:
             expires_in = Config.instance().settings.Controller.jwt_access_token_expire_minutes
         return self._create_token(username, token_version, "access", expires_in, secret_key)
 
     def create_refresh_token(
-        self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0
+        self, username, token_version: int = 0, secret_key: Optional[str] = None, expires_in: int = 0
     ) -> str:
 
         if not expires_in:
             expires_in = Config.instance().settings.Controller.jwt_refresh_token_expire_minutes
         return self._create_token(username, token_version, "refresh", expires_in, secret_key)
 
-    def get_token_data(self, token: str, secret_key: str = None) -> TokenData:
+    def get_token_data(self, token: str, secret_key: Optional[str] = None) -> TokenData:
 
         def auth_error(detail: str) -> HTTPException:
             return HTTPException(
@@ -105,7 +107,7 @@ class AuthService:
         key = OctKey.import_key(secret_key)
         try:
             payload = jwt.decode(token, key, algorithms=[algorithm])
-            username: str = payload.claims.get("sub")
+            username: Optional[str] = payload.claims.get("sub")
             if username is None:
                 raise auth_error("Invalid token: missing subject claim")
             # Validate the exp claim — joserfc does not validate time-based claims by default
@@ -123,5 +125,5 @@ class AuthService:
             raise auth_error(f"Invalid token ({type(e).__name__})")
         return token_data
 
-    def get_username_from_token(self, token: str, secret_key: str = None) -> Optional[str]:
+    def get_username_from_token(self, token: str, secret_key: Optional[str] = None) -> Optional[str]:
         return self.get_token_data(token, secret_key).username

gns3server/services/computes.py

@@ -43,7 +43,7 @@ class ComputesService:
 
     async def create_compute(self, compute_create: schemas.ComputeCreate, connect: bool = False) -> models.Compute:
 
-        if await self._computes_repo.get_compute(compute_create.compute_id):
+        if compute_create.compute_id and await self._computes_repo.get_compute(compute_create.compute_id):
             raise ControllerBadRequestError(f"Compute '{compute_create.compute_id}' is already registered")
         db_compute = await self._computes_repo.create_compute(compute_create)
         compute = await self._controller.add_compute(

gns3server/services/templates.py

@@ -20,7 +20,7 @@ import pydantic
 
 from uuid import UUID
 from fastapi.encoders import jsonable_encoder
-from typing import List
+from typing import List, Optional
 
 from gns3server import schemas
 from gns3server.config import Config
@@ -34,7 +34,7 @@ from gns3server.controller.controller_error import (
 )
 
 
-TEMPLATE_TYPE_TO_SCHEMA = {
+TEMPLATE_TYPE_TO_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
     "cloud": schemas.CloudTemplate,
     "ethernet_hub": schemas.EthernetHubTemplate,
     "ethernet_switch": schemas.EthernetSwitchTemplate,
@@ -47,7 +47,7 @@ TEMPLATE_TYPE_TO_SCHEMA = {
     "qemu": schemas.QemuTemplate,
 }
 
-TEMPLATE_TYPE_TO_UPDATE_SCHEMA = {
+TEMPLATE_TYPE_TO_UPDATE_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
     "cloud": schemas.CloudTemplateUpdate,
     "ethernet_hub": schemas.EthernetHubTemplateUpdate,
     "ethernet_switch": schemas.EthernetSwitchTemplateUpdate,
@@ -59,7 +59,7 @@ TEMPLATE_TYPE_TO_UPDATE_SCHEMA = {
     "qemu": schemas.QemuTemplateUpdate,
 }
 
-DYNAMIPS_PLATFORM_TO_SCHEMA = {
+DYNAMIPS_PLATFORM_TO_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
     "c7200": schemas.C7200DynamipsTemplate,
     "c3745": schemas.C3745DynamipsTemplate,
     "c3725": schemas.C3725DynamipsTemplate,
@@ -69,7 +69,7 @@ DYNAMIPS_PLATFORM_TO_SCHEMA = {
     "c1700": schemas.C1700DynamipsTemplate,
 }
 
-DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA = {
+DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
     "c7200": schemas.C7200DynamipsTemplateUpdate,
     "c3745": schemas.C3745DynamipsTemplateUpdate,
     "c3725": schemas.C3725DynamipsTemplateUpdate,
@@ -169,11 +169,12 @@ class TemplatesService:
         for builtin_template in BUILTIN_TEMPLATES:
             builtin_template["symbol"] = self._controller.symbols.resolve_symbol(builtin_template["symbol"])
 
-    def get_builtin_template(self, template_id: UUID) -> dict:
+    def get_builtin_template(self, template_id: UUID) -> Optional[dict]:
 
         for builtin_template in BUILTIN_TEMPLATES:
             if builtin_template["template_id"] == template_id:
                 return jsonable_encoder(builtin_template)
+        return None
 
     def _base_path(self):
         return self._templates_repo.configs_path()
@@ -304,7 +305,7 @@ class TemplatesService:
         try:
             # validate the update settings
             update_settings = jsonable_encoder(template_update, exclude_unset=True)
-            if db_template.template_type == "dynamips":
+            if isinstance(db_template, models.DynamipsTemplate):
                 template_schema = DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA[db_template.platform]
             else:
                 template_schema = TEMPLATE_TYPE_TO_UPDATE_SCHEMA[db_template.template_type]
@@ -313,9 +314,9 @@ class TemplatesService:
             raise ControllerBadRequestError(f"JSON schema error received while updating template: {e}")
 
         images_to_add_to_template = await self._find_images(db_template.template_type, template_settings)
-        if db_template.template_type == "dynamips" and "image" in template_settings:
+        if isinstance(db_template, models.DynamipsTemplate) and "image" in template_settings:
             await self._remove_image(db_template.template_id, db_template.image)
-        elif db_template.template_type == "iou" and "path" in template_settings:
+        elif isinstance(db_template, models.IOUTemplate) and "path" in template_settings:
             await self._remove_image(db_template.template_id, db_template.path)
         elif db_template.template_type == "qemu":
             for key in template_update.model_dump().keys():