gns3-server/gns3server/schemas/config.py

#
# Copyright (C) 2021 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import socket

from enum import Enum
from pydantic import (
    ConfigDict,
    BaseModel,
    Field,
    SecretStr,
    FilePath,
    DirectoryPath,
    field_validator,
    model_validator
)
from typing import List


class ControllerSettings(BaseModel):

    jwt_secret_key: str = None
    jwt_algorithm: str = "HS256"
    jwt_access_token_expire_minutes: int = 1440  # 24 hours
    jwt_refresh_token_expire_minutes: int = 43200  # 30 days
    default_admin_username: str = "admin"
    default_admin_password: SecretStr = SecretStr("admin")
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class VPCSSettings(BaseModel):

    vpcs_path: str = "vpcs"
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class DynamipsSettings(BaseModel):

    allocate_aux_console_ports: bool = False
    mmap_support: bool = True
    dynamips_path: str = "dynamips"
    sparse_memory_support: bool = True
    ghost_ios_support: bool = True
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class IOUSettings(BaseModel):

    iourc_path: str = None
    license_check: bool = True
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class QemuSettings(BaseModel):

    enable_monitor: bool = True
    monitor_host: str = "127.0.0.1"
    enable_hardware_acceleration: bool = True
    require_hardware_acceleration: bool = False
    allow_unsafe_options: bool = False
    ovmf_firmware_dir: str = "/usr/share/OVMF"
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class VirtualBoxSettings(BaseModel):

    vboxmanage_path: str = None
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class VMwareSettings(BaseModel):

    vmrun_path: str = None
    vmnet_start_range: int = Field(2, ge=1, le=255)
    vmnet_end_range: int = Field(255, ge=1, le=255)  # should be limited to 19 on Windows
    block_host_traffic: bool = False
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)

    @model_validator(mode="after")
    def check_vmnet_port_range(self) -> "VMwareSettings":
        if self.vmnet_end_range <= self.vmnet_start_range:
            raise ValueError("vmnet_end_range must be > vmnet_start_range")
        return self


class WebWiresharkSettings(BaseModel):

    enabled: bool = True
    image: str = "gns3/web-wireshark:latest"
    network_subnet: str = "172.31.0.0/22"
    memory: str = "2g"
    cpus: float = 1.0
    pids_limit: int = 1000
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class ServerProtocol(str, Enum):

    http = "http"
    https = "https"


class UbridgeControlTransport(str, Enum):

    # TCP control channel: -H host:port. ubridge now binds loopback by default,
    # so this is reachable only locally. Retained for backward compatibility.
    tcp = "tcp"
    # AF_UNIX control channel: -U socket_path, authenticated in-kernel via
    # SO_PEERCRED (ubridge accepts only its own UID). Recommended on Linux.
    unix = "unix"


class BuiltinSymbolTheme(str, Enum):

    classic = "Classic"
    affinity_square_blue = "Affinity-square-blue"
    affinity_square_red = "Affinity-square-red"
    affinity_square_gray = "Affinity-square-gray"
    affinity_circle_blue = "Affinity-circle-blue"
    affinity_circle_red = "Affinity-circle-red"
    affinity_circle_gray = "Affinity-circle-gray"


class ServerSettings(BaseModel):

    local: bool = False
    enable_http_auth: bool = True
    name: str = f"{socket.gethostname()} (controller)"
    protocol: ServerProtocol = ServerProtocol.http
    host: str = "0.0.0.0"
    port: int = Field(3080, gt=0, le=65535)
    secrets_dir: DirectoryPath = None
    certfile: FilePath = None
    certkey: FilePath = None
    enable_ssl: bool = False
    images_path: str = "~/GNS3/images"
    projects_path: str = "~/GNS3/projects"
    appliances_path: str = "~/GNS3/appliances"
    symbols_path: str = "~/GNS3/symbols"
    configs_path: str = "~/GNS3/configs"
    resources_path: str = None
    default_symbol_theme: BuiltinSymbolTheme = BuiltinSymbolTheme.affinity_square_blue
    allow_raw_images: bool = True
    auto_discover_images: bool = True
    report_errors: bool = True
    additional_images_paths: List[str] = Field(default_factory=list)
    console_start_port_range: int = Field(5000, gt=0, le=65535)
    console_end_port_range: int = Field(10000, gt=0, le=65535)
    vnc_console_start_port_range: int = Field(5900, ge=5900, le=65535)
    vnc_console_end_port_range: int = Field(10000, ge=5900, le=65535)
    udp_start_port_range: int = Field(10000, gt=0, le=65535)
    udp_end_port_range: int = Field(30000, gt=0, le=65535)
    ubridge_path: str = "ubridge"
    # Transport for the uBridge hypervisor control channel. "unix" (-U,
    # AF_UNIX + SO_PEERCRED) is the default — recommended on Linux for
    # kernel-level peer authentication. "tcp" (-H) is retained for backward
    # compatibility.
    ubridge_control_transport: UbridgeControlTransport = UbridgeControlTransport.unix
    # Marker (traffic-insight) UDP sink: one listener per compute process that
    # receives ubridge MARK signals from every ubridge on this host. The host
    # defaults to loopback because ubridge runs on the same host as the compute.
    # port=0 lets the OS choose a free port (read back and handed to ubridge).
    marker_listen_host: str = "127.0.0.1"
    marker_listen_port: int = Field(3070, ge=0, le=65535)
    compute_username: str = "gns3"
    compute_password: SecretStr = SecretStr("")
    allowed_interfaces: List[str] = Field(default_factory=list)
    default_nat_interface: str = None
    allow_remote_console: bool = False
    enable_builtin_templates: bool = True
    install_builtin_appliances: bool = True
    skills_repo_url: str = "https://github.com/gns3/gns3-skills.git"
    skills_repo_branch: str = "main"
    skills_auto_update: bool = True

    # MCP (Model Context Protocol) transport security settings
    # DNS rebinding protection is disabled by default to allow connections
    # from any host (aligns with GNS3 server's 0.0.0.0 binding).
    # Users with security requirements can enable protection and specify
    # allowed hosts using "host:*" port wildcard patterns.
    mcp_enable_dns_rebinding_protection: bool = False
    mcp_allowed_hosts: list[str] = Field(default_factory=list)
    mcp_allowed_origins: list[str] = Field(default_factory=list)

    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)

    @field_validator("mcp_allowed_hosts", mode="before")
    @classmethod
    def split_mcp_allowed_hosts(cls, v):
        if v and isinstance(v, str):
            return v.split(",")
        if not v:
            return list()
        return v

    @field_validator("mcp_allowed_origins", mode="before")
    @classmethod
    def split_mcp_allowed_origins(cls, v):
        if v and isinstance(v, str):
            return v.split(",")
        if not v:
            return list()
        return v

    @field_validator("additional_images_paths", mode="before")
    @classmethod
    def split_additional_images_paths(cls, v):
        if v and isinstance(v, str):
            return v.split(";")
        if not v:
            return list()
        return v

    @field_validator("allowed_interfaces", mode="before")
    @classmethod
    def split_allowed_interfaces(cls, v):
        if v and isinstance(v, str):
            return v.split(",")
        if not v:
            return list()
        return v

    @model_validator(mode="after")
    def check_console_port_range(self) -> "ServerSettings":
        if self.console_end_port_range <= self.console_start_port_range:
            raise ValueError("console_end_port_range must be > console_start_port_range")
        return self

    @model_validator(mode="after")
    def check_vnc_port_range(self) -> "ServerSettings":
        if self.vnc_console_end_port_range <= self.vnc_console_start_port_range:
            raise ValueError("vnc_console_end_port_range must be > vnc_console_start_port_range")
        return self

    @model_validator(mode="after")
    def check_enable_ssl(self) -> "ServerSettings":
        if self.enable_ssl is True:
            if not self.certfile:
                raise ValueError("SSL is enabled but certfile is not configured")
            if not self.certkey:
                raise ValueError("SSL is enabled but certkey is not configured")
        return self


class ServerConfig(BaseModel):

    Server: ServerSettings = ServerSettings()
    Controller: ControllerSettings = ControllerSettings()
    VPCS: VPCSSettings = VPCSSettings()
    Dynamips: DynamipsSettings = DynamipsSettings()
    IOU: IOUSettings = IOUSettings()
    Qemu: QemuSettings = QemuSettings()
    VirtualBox: VirtualBoxSettings = VirtualBoxSettings()
    VMware: VMwareSettings = VMwareSettings()
    WebWireshark: WebWiresharkSettings = WebWiresharkSettings()