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, Optional


class ControllerSettings(BaseModel):
    jwt_secret_key: Optional[str] = Field(
        None,
        description="Secret key used to sign the JWT authentication tokens "
        "(normally managed via the secrets directory, not the configuration file)",
    )
    jwt_algorithm: str = Field("HS256", description="Algorithm used to sign the JWT tokens")
    jwt_access_token_expire_minutes: int = Field(
        1440, description="Lifetime of the JWT access tokens in minutes (24 hours by default)"
    )
    jwt_refresh_token_expire_minutes: int = Field(
        43200, description="Lifetime of the JWT refresh tokens in minutes (30 days by default)"
    )
    default_admin_username: str = Field(
        "admin",
        description="Username of the super admin account seeded when the controller database is created; "
        "changing it has no effect until the database is re-created (which resets the account)",
    )
    default_admin_password: SecretStr = Field(
        SecretStr("admin"),
        description="Password of the super admin account seeded when the controller database is created; "
        "changing it has no effect until the database is re-created (which resets the account)",
    )
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class VPCSSettings(BaseModel):
    vpcs_path: str = Field("vpcs", description="VPCS executable location, default: search in PATH")
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class DynamipsSettings(BaseModel):
    allocate_aux_console_ports: bool = Field(False, description="Allocate auxiliary console ports on IOS routers")
    mmap_support: bool = Field(
        True, description="Use memory-mapped flash files (mmap) to lower the memory usage of routers"
    )
    dynamips_path: str = Field("dynamips", description="Dynamips executable location, default: search in PATH")
    sparse_memory_support: bool = Field(
        True, description="Use sparse memory allocation to lower the memory usage of routers"
    )
    ghost_ios_support: bool = Field(
        True, description="Enable Ghost IOS support to share memory between identical IOS images"
    )
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class IOUSettings(BaseModel):
    iourc_path: Optional[str] = Field(
        None, description="Path of your .iourc file, the file is searched in $HOME/.iourc if not provided"
    )
    license_check: bool = Field(
        True,
        description="Validate the iourc license file (if disabled, IOU will not start and no errors "
        "will be shown when the license is invalid)",
    )
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class QemuSettings(BaseModel):
    enable_monitor: bool = Field(True, description="Use the Qemu monitor feature to communicate with Qemu VMs")
    monitor_host: str = Field("127.0.0.1", description="IP used to listen for the monitor")
    enable_hardware_acceleration: bool = Field(True, description="Enable hardware acceleration (KVM)")
    require_hardware_acceleration: bool = Field(
        False, description="Require hardware acceleration in order to start VMs"
    )
    allow_unsafe_options: bool = Field(False, description="Allow unsafe additional command line options")
    ovmf_firmware_dir: str = Field("/usr/share/OVMF", description="Path to the OVMF firmware directory")
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class VirtualBoxSettings(BaseModel):
    vboxmanage_path: Optional[str] = None
    model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)


class VMwareSettings(BaseModel):
    vmrun_path: Optional[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 = Field(
        True, description="Enable the Web Wireshark feature (container-based Wireshark in the browser)"
    )
    image: str = Field("gns3/web-wireshark:latest", description="Docker image for the Web Wireshark containers")
    network_subnet: str = Field(
        "172.31.0.0/22",
        description="Docker network subnet for the Web Wireshark containers (change it if it conflicts "
        "with your existing network)",
    )
    memory: str = Field("2g", description='Memory limit per container (e.g. "512m", "2g")')
    cpus: float = Field(1.0, description="CPU cores per container (e.g. 1.0, 2.0)")
    pids_limit: int = Field(1000, description="Process limit per container")
    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 = Field(
        False, description="Local server mode, set by the --local command line argument (not meant to be set by hand)"
    )
    enable_http_auth: bool = Field(True, description="Enable compute HTTP authentication")
    name: str = Field(
        f"{socket.gethostname()} (controller)",
        description="Server name, default is what is returned by socket.gethostname()",
    )
    protocol: ServerProtocol = Field(ServerProtocol.http, description="Protocol used by the server: http or https")
    host: str = Field("0.0.0.0", description="IP address where the server listens for connections")
    port: int = Field(3080, gt=0, le=65535, description="HTTP port used to control the server")
    secrets_dir: Optional[DirectoryPath] = Field(
        None, description="Directory where secrets are stored (e.g. the JWT secret key)"
    )
    certfile: Optional[FilePath] = Field(None, description="SSL certificate file, requires enable_ssl")
    certkey: Optional[FilePath] = Field(None, description="SSL key file, requires enable_ssl")
    enable_ssl: bool = Field(False, description="Enable SSL encryption")
    images_path: str = Field("~/GNS3/images", description="Path where binary images are stored")
    projects_path: str = Field("~/GNS3/projects", description="Path where user projects are stored")
    appliances_path: str = Field("~/GNS3/appliances", description="Path where custom user appliances are stored")
    symbols_path: str = Field("~/GNS3/symbols", description="Path where custom user symbols are stored")
    configs_path: str = Field("~/GNS3/configs", description="Path where custom user configs are stored")
    resources_path: Optional[str] = Field(
        None,
        description="Path where files like built-in appliances and Docker resources are stored "
        "(defaults to the local user data directory)",
    )
    default_symbol_theme: BuiltinSymbolTheme = Field(
        BuiltinSymbolTheme.affinity_square_blue,
        description='Default symbol theme, e.g. "Classic" or "Affinity-square-blue"',
    )
    allow_raw_images: bool = Field(True, description="Allow raw images to be uploaded to the server")
    auto_discover_images: bool = Field(True, description="Automatically discover images in the images directory")
    image_sync_interval: int = Field(900, ge=10, description="Seconds between automatic image inventory scans")
    report_errors: bool = Field(True, description="Automatically send crash reports to the GNS3 team")
    additional_images_paths: List[str] = Field(
        default_factory=list,
        description="Additional paths to look for images (semicolon-separated in the configuration file)",
    )
    console_start_port_range: int = Field(
        5000, gt=0, le=65535, description="First console port of the range allocated to devices"
    )
    console_end_port_range: int = Field(
        10000, gt=0, le=65535, description="Last console port of the range allocated to devices"
    )
    vnc_console_start_port_range: int = Field(
        5900, ge=5900, le=65535, description="First VNC console port of the range allocated to devices"
    )
    vnc_console_end_port_range: int = Field(
        10000, ge=5900, le=65535, description="Last VNC console port of the range allocated to devices"
    )
    udp_start_port_range: int = Field(
        10000,
        gt=0,
        le=65535,
        description="First UDP port of the range allocated for inter-device communication (two ports per link)",
    )
    udp_end_port_range: int = Field(
        30000,
        gt=0,
        le=65535,
        description="Last UDP port of the range allocated for inter-device communication (two ports per link)",
    )
    ubridge_path: str = Field("ubridge", description="uBridge executable location, default: search in PATH")
    ubridge_control_transport: UbridgeControlTransport = Field(
        UbridgeControlTransport.unix,
        description='uBridge control channel transport: "unix" (AF_UNIX + SO_PEERCRED, recommended '
        'on Linux) or "tcp" (loopback, kept for backward compatibility)',
    )
    marker_listen_host: str = Field(
        "127.0.0.1",
        description="Marker (traffic-insight) UDP sink listen host: one listener per compute process "
        "receives uBridge MARK signals from every uBridge on this host",
    )
    marker_listen_port: int = Field(
        3070, ge=0, le=65535, description="Marker UDP sink listen port (0 lets the operating system choose a free port)"
    )
    compute_username: str = Field("gns3", description='Username for compute HTTP authentication, "gns3" is the default')
    compute_password: SecretStr = Field(
        SecretStr(""),
        description="Password for compute HTTP authentication, a randomly generated password is used if not set",
    )
    allowed_interfaces: List[str] = Field(
        default_factory=list,
        description="Only allow these interfaces to be used by GNS3, for the Cloud node for example "
        "(comma-separated; do not forget virbr0 for the NAT node to work)",
    )
    default_nat_interface: Optional[str] = Field(
        None, description="Interface used by the NAT node, default is virbr0 on Linux (requires libvirt)"
    )
    allow_remote_console: bool = Field(
        False,
        description="Allow console connections from remote machines "
        "(console ports only accept local connections by default)",
    )
    enable_builtin_templates: bool = Field(True, description="Enable the built-in templates")
    install_builtin_appliances: bool = Field(True, description="Install the built-in appliances")
    skills_repo_url: str = Field(
        "https://github.com/gns3/gns3-skills.git",
        description="Git repository URL for the external GNS3 Copilot skills "
        "(injection skills, prompts and device skills)",
    )
    skills_repo_branch: str = Field("main", description="Git branch of the skills repository")
    skills_auto_update: bool = Field(
        True, description="Automatically pull updates from the skills repository when reloading"
    )
    mcp_enable_dns_rebinding_protection: bool = Field(
        False,
        description="Enable MCP transport DNS rebinding protection (allowed hosts and origins must be configured)",
    )
    mcp_allowed_hosts: list[str] = Field(
        default_factory=list,
        description='Allowed hosts for MCP connections, only "host:*" port wildcards are supported '
        '(e.g. "127.0.0.1:*")',
    )
    mcp_allowed_origins: list[str] = Field(
        default_factory=list, description='Allowed origins for MCP connections (e.g. "http://localhost:*")'
    )

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