ExternalVendorCode/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2024-12-18 20:37:57 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Rename ssl and auth configuration file settings.

Browse Source 
Add enable SSL config validator.
Strict configuration file validation: any error will prevent the server to start.
Core server logic moved to a Server class.
This commit is contained in:
grossmj 2021-04-12 23:26:42 +09:30
parent 1b5a5de4bc
commit 71725aade6
12 changed files with 542 additions and 520 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -8,6 +8,9 @@ __pycache__
# environment file
.env
# hypothesis files
.hypothesis
# C extensions
*.so
@ -62,4 +65,5 @@ startup.vpcs
# Virtualenv
env
venv
*venv
.ropeproject

5
conf/gns3_server.conf
View File

@ -12,7 +12,7 @@ port = 3080
secrets_dir = /home/gns3/.config/GNS3/secrets
; Options to enable SSL encryption
ssl = False
enable_ssl = False
certfile = /home/gns3/.config/GNS3/ssl/server.cert
certkey = /home/gns3/.config/GNS3/ssl/server.key
@ -58,7 +58,7 @@ udp_end_port_range = 30000
;ubridge_path = ubridge
; Option to enable HTTP authentication.
auth = False
enable_http_auth = False
; Username for HTTP authentication.
user = gns3
; Password for HTTP authentication.
@ -78,7 +78,6 @@ jwt_secret_key = efd08eccec3bd0a1be2e086670e5efa90969c68d07e072d7354a76cea5e33d4
jwt_algorithm = HS256
jwt_access_token_expire_minutes = 1440
[VPCS]
; VPCS executable location, default: search in PATH
;vpcs_path = vpcs

6
gns3server/config.py
View File

@ -272,14 +272,14 @@ class Config:
return
for file in parsed_files:
log.info(f"Load configuration file '{file}'")
log.info(f"Configuration file '{file}' loaded")
self._watched_files[file] = os.stat(file).st_mtime
try:
self._settings = ServerConfig(**config._sections)
except ValidationError as e:
log.error(f"Could not validate config: {e}")
return
log.critical(f"Could not validate configuration file settings: {e}")
raise
self._load_secret_files()

6
gns3server/controller/__init__.py
View File

@ -81,7 +81,7 @@ class Controller:
self._load_controller_settings()
if server_config.ssl:
if server_config.enable_ssl:
if sys.platform.startswith("win"):
log.critical("SSL mode is not supported on Windows")
raise SystemExit
@ -242,11 +242,11 @@ class Controller:
iou_config = Config.instance().settings.IOU
server_config = Config.instance().settings.Server
#controller_config.getboolean("iou_license_check", True)
if iou_config.iourc_path:
iourc_path = iou_config.iourc_path
else:
if not server_config.secrets_dir:
server_config.secrets_dir = os.path.dirname(Config.instance().server_config)
iourc_path = os.path.join(server_config.secrets_dir, "gns3_iourc_license")
if os.path.exists(iourc_path):

4
gns3server/main.py
View File

@ -80,8 +80,8 @@ def main():
if not sys.platform.startswith("win"):
if "--daemon" in sys.argv:
daemonize()
from gns3server.run import run
run()
from gns3server.server import Server
Server().run()
if __name__ == '__main__':

352
gns3server/run.py
View File

@ -1,352 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 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/>.
"""
Start the program. Use main.py to load it.
"""
import os
import datetime
import locale
import argparse
import psutil
import sys
import asyncio
import signal
import functools
import uvicorn
import secrets
from gns3server.controller import Controller
from gns3server.compute.port_manager import PortManager
from gns3server.logger import init_logger
from gns3server.version import __version__
from gns3server.config import Config
from gns3server.crash_report import CrashReport
from gns3server.api.server import app
import logging
log = logging.getLogger(__name__)
def locale_check():
"""
Checks if this application runs with a correct locale (i.e. supports UTF-8 encoding) and attempt to fix
if this is not the case.
This is to prevent UnicodeEncodeError with unicode paths when using standard library I/O operation
methods (e.g. os.stat() or os.path.*) which rely on the system or user locale.
More information can be found there: http://seasonofcode.com/posts/unicode-i-o-and-locales-in-python.html
or there: http://robjwells.com/post/61198832297/get-your-us-ascii-out-of-my-face
"""
# no need to check on Windows or when this application is frozen
if sys.platform.startswith("win") or hasattr(sys, "frozen"):
return
language = encoding = None
try:
language, encoding = locale.getlocale()
except ValueError as e:
log.error("Could not determine the current locale: {}".format(e))
if not language and not encoding:
try:
log.warning("Could not find a default locale, switching to C.UTF-8...")
locale.setlocale(locale.LC_ALL, ("C", "UTF-8"))
except locale.Error as e:
log.error("Could not switch to the C.UTF-8 locale: {}".format(e))
raise SystemExit
elif encoding != "UTF-8":
log.warning("Your locale {}.{} encoding is not UTF-8, switching to the UTF-8 version...".format(language, encoding))
try:
locale.setlocale(locale.LC_ALL, (language, "UTF-8"))
except locale.Error as e:
log.error("Could not set an UTF-8 encoding for the {} locale: {}".format(language, e))
raise SystemExit
else:
log.info("Current locale is {}.{}".format(language, encoding))
def parse_arguments(argv):
"""
Parse command line arguments and override local configuration
:params args: Array of command line arguments
"""
parser = argparse.ArgumentParser(description="GNS3 server version {}".format(__version__))
parser.add_argument("-v", "--version", help="show the version", action="version", version=__version__)
parser.add_argument("--host", help="run on the given host/IP address")
parser.add_argument("--port", help="run on the given port", type=int)
parser.add_argument("--ssl", action="store_true", help="run in SSL mode")
parser.add_argument("--config", help="Configuration file")
parser.add_argument("--certfile", help="SSL cert file")
parser.add_argument("--certkey", help="SSL key file")
parser.add_argument("-L", "--local", action="store_true", help="local mode (allows some insecure operations)")
parser.add_argument("-A", "--allow", action="store_true", help="allow remote connections to local console ports")
parser.add_argument("-q", "--quiet", action="store_true", help="do not show logs on stdout")
parser.add_argument("-d", "--debug", action="store_true", help="show debug logs")
parser.add_argument("--log", help="send output to logfile instead of console")
parser.add_argument("--logmaxsize", help="maximum logfile size in bytes (default is 10MB)")
parser.add_argument("--logbackupcount", help="number of historical log files to keep (default is 10)")
parser.add_argument("--logcompression", action="store_true", help="compress inactive (historical) logs")
parser.add_argument("--daemon", action="store_true", help="start as a daemon")
parser.add_argument("--pid", help="store process pid")
parser.add_argument("--profile", help="Settings profile (blank will use default settings files)")
args = parser.parse_args(argv)
if args.config:
Config.instance(files=[args.config], profile=args.profile)
else:
Config.instance(profile=args.profile)
config = Config.instance().settings
defaults = {
"host": config.Server.host,
"port": config.Server.port,
"ssl": config.Server.ssl,
"certfile": config.Server.certfile,
"certkey": config.Server.certkey,
"local": config.Server.local,
"allow": config.Server.allow_remote_console,
"quiet": config.Server.quiet,
"debug": config.Server.debug,
"logfile": config.Server.logfile,
"logmaxsize": config.Server.logmaxsize,
"logbackupcount": config.Server.logbackupcount,
"logcompression": config.Server.logcompression
}
parser.set_defaults(**defaults)
return parser.parse_args(argv)
def set_config(args):
config = Config.instance().settings
config.Server.local = args.local
config.Server.allow_remote_console = args.allow
config.Server.host = args.host
config.Server.port = args.port
config.Server.ssl = args.ssl
config.Server.certfile = args.certfile
config.Server.certkey = args.certkey
config.Server.debug = args.debug
def pid_lock(path):
"""
Write the file in a file on the system.
Check if the process is not already running.
"""
if os.path.exists(path):
pid = None
try:
with open(path) as f:
try:
pid = int(f.read())
os.kill(pid, 0) # kill returns an error if the process is not running
except (OSError, SystemError, ValueError):
pid = None
except OSError as e:
log.critical("Can't open pid file %s: %s", pid, str(e))
sys.exit(1)
if pid:
log.critical("GNS3 is already running pid: %d", pid)
sys.exit(1)
try:
with open(path, 'w+') as f:
f.write(str(os.getpid()))
except OSError as e:
log.critical("Can't write pid file %s: %s", path, str(e))
sys.exit(1)
def kill_ghosts():
"""
Kill process from previous GNS3 session
"""
detect_process = ["vpcs", "traceng", "ubridge", "dynamips"]
for proc in psutil.process_iter():
try:
name = proc.name().lower().split(".")[0]
if name in detect_process:
proc.kill()
log.warning("Killed ghost process %s", name)
except (OSError, psutil.NoSuchProcess, psutil.AccessDenied):
pass
async def reload_server():
"""
Reload the server.
"""
await Controller.instance().reload()
def signal_handling():
def signal_handler(signame, *args):
try:
if signame == "SIGHUP":
log.info("Server has got signal {}, reloading...".format(signame))
asyncio.ensure_future(reload_server())
else:
log.info("Server has got signal {}, exiting...".format(signame))
os.kill(os.getpid(), signal.SIGTERM)
except asyncio.CancelledError:
pass
signals = [] # SIGINT and SIGTERM are already registered by uvicorn
if sys.platform.startswith("win"):
signals.extend(["SIGBREAK"])
else:
signals.extend(["SIGHUP", "SIGQUIT"])
for signal_name in signals:
callback = functools.partial(signal_handler, signal_name)
if sys.platform.startswith("win"):
# add_signal_handler() is not yet supported on Windows
signal.signal(getattr(signal, signal_name), callback)
else:
loop = asyncio.get_event_loop()
loop.add_signal_handler(getattr(signal, signal_name), callback)
def run():
args = parse_arguments(sys.argv[1:])
if args.daemon and sys.platform.startswith("win"):
log.critical("Daemon is not supported on Windows")
sys.exit(1)
if args.pid:
pid_lock(args.pid)
kill_ghosts()
level = logging.INFO
if args.debug:
level = logging.DEBUG
stream_handler = init_logger(level,
logfile=args.log,
max_bytes=int(args.logmaxsize),
backup_count=int(args.logbackupcount),
compression=args.logcompression,
quiet=args.quiet)
log.info("GNS3 server version {}".format(__version__))
current_year = datetime.date.today().year
log.info("Copyright (c) 2007-{} GNS3 Technologies Inc.".format(current_year))
for config_file in Config.instance().get_config_files():
log.info("Config file {} loaded".format(config_file))
set_config(args)
config = Config.instance().settings
if config.Server.local:
log.warning("Local mode is enabled. Beware, clients will have full control on your filesystem")
if config.Server.auth:
log.info("HTTP authentication is enabled with username '{}'".format(config.Server.user))
# we only support Python 3 version >= 3.6
if sys.version_info < (3, 6, 0):
raise SystemExit("Python 3.6 or higher is required")
log.info("Running with Python {major}.{minor}.{micro} and has PID {pid}".format(major=sys.version_info[0],
minor=sys.version_info[1],
micro=sys.version_info[2],
pid=os.getpid()))
# check for the correct locale (UNIX/Linux only)
locale_check()
try:
os.getcwd()
except FileNotFoundError:
log.critical("The current working directory doesn't exist")
return
CrashReport.instance()
host = config.Server.host
port = config.Server.port
PortManager.instance().console_host = host
signal_handling()
try:
log.info("Starting server on {}:{}".format(host, port))
# only show uvicorn access logs in debug mode
access_log = False
if log.getEffectiveLevel() == logging.DEBUG:
access_log = True
if config.Server.ssl:
if sys.platform.startswith("win"):
log.critical("SSL mode is not supported on Windows")
raise SystemExit
log.info("SSL is enabled")
config = uvicorn.Config(app,
host=host,
port=port,
access_log=access_log,
ssl_certfile=config.Server.certfile,
ssl_keyfile=config.Server.certkey,
lifespan="on")
# overwrite uvicorn loggers with our own logger
for uvicorn_logger_name in ("uvicorn", "uvicorn.error"):
uvicorn_logger = logging.getLogger(uvicorn_logger_name)
uvicorn_logger.handlers = [stream_handler]
uvicorn_logger.propagate = False
if access_log:
uvicorn_logger = logging.getLogger("uvicorn.access")
uvicorn_logger.handlers = [stream_handler]
uvicorn_logger.propagate = False
server = uvicorn.Server(config)
loop = asyncio.get_event_loop()
loop.run_until_complete(server.serve())
except OSError as e:
# This is to ignore OSError: [WinError 0] The operation completed successfully exception on Windows.
if not sys.platform.startswith("win") or not e.winerror == 0:
raise
except Exception as e:
log.critical("Critical error while running the server: {}".format(e), exc_info=1)
CrashReport.instance().capture_exception()
return
finally:
if args.pid:
log.info("Remove PID file %s", args.pid)
try:
os.remove(args.pid)
except OSError as e:
log.critical("Can't remove pid file %s: %s", args.pid, str(e))

31
gns3server/schemas/config.py
View File

@ -16,7 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from enum import Enum
from pydantic import BaseModel, Field, SecretStr, validator
from pydantic import BaseModel, Field, SecretStr, FilePath, validator
from typing import List
@ -110,13 +110,14 @@ class ServerProtocol(str, Enum):
class ServerSettings(BaseModel):
local: bool = False
protocol: ServerProtocol = ServerProtocol.http
host: str = "0.0.0.0"
port: int = Field(3080, gt=0, le=65535)
secrets_dir: str = None
ssl: bool = False
certfile: str = None
certkey: str = 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"
@ -133,18 +134,10 @@ class ServerSettings(BaseModel):
ubridge_path: str = "ubridge"
user: str = None
password: SecretStr = None
auth: bool = False
enable_http_auth: bool = False
allowed_interfaces: List[str] = Field(default_factory=list)
default_nat_interface: str = None
logfile: str = None
logmaxsize: int = 10000000 # default is 10MB
logbackupcount: int = 10
logcompression: bool = False
local: bool = False
allow_remote_console: bool = False
quiet: bool = False
debug: bool = False
@validator("additional_images_paths", pre=True)
def split_additional_images_paths(cls, v):
@ -170,7 +163,7 @@ class ServerSettings(BaseModel):
raise ValueError("vnc_console_end_port_range must be > vnc_console_start_port_range")
return v
@validator("auth")
@validator("enable_http_auth")
def validate_enable_auth(cls, v, values):
if v is True:
@ -178,6 +171,16 @@ class ServerSettings(BaseModel):
raise ValueError("HTTP authentication is enabled but no username is configured")
return v
@validator("enable_ssl")
def validate_enable_ssl(cls, v, values):
if v is True:
if "certfile" not in values or not values["certfile"]:
raise ValueError("SSL is enabled but certfile is not configured")
if "certkey" not in values or not values["certkey"]:
raise ValueError("SSL is enabled but certkey is not configured")
return v
class Config:
validate_assignment = True
anystr_strip_whitespace = True

355
gns3server/server.py Normal file
View File

@ -0,0 +1,355 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# 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/>.
"""
Start the program. Use main.py to load it.
"""
import os
import datetime
import locale
import argparse
import psutil
import sys
import asyncio
import signal
import functools
import uvicorn
from gns3server.controller import Controller
from gns3server.compute.port_manager import PortManager
from gns3server.logger import init_logger
from gns3server.version import __version__
from gns3server.config import Config
from gns3server.crash_report import CrashReport
from gns3server.api.server import app
from pydantic import ValidationError
import logging
log = logging.getLogger(__name__)
class Server:
_stream_handler = None
@staticmethod
def _locale_check():
"""
Checks if this application runs with a correct locale (i.e. supports UTF-8 encoding) and attempt to fix
if this is not the case.
This is to prevent UnicodeEncodeError with unicode paths when using standard library I/O operation
methods (e.g. os.stat() or os.path.*) which rely on the system or user locale.
More information can be found there: http://seasonofcode.com/posts/unicode-i-o-and-locales-in-python.html
or there: http://robjwells.com/post/61198832297/get-your-us-ascii-out-of-my-face
"""
# no need to check on Windows or when this application is frozen
if sys.platform.startswith("win") or hasattr(sys, "frozen"):
return
language = encoding = None
try:
language, encoding = locale.getlocale()
except ValueError as e:
log.error("Could not determine the current locale: {}".format(e))
if not language and not encoding:
try:
log.warning("Could not find a default locale, switching to C.UTF-8...")
locale.setlocale(locale.LC_ALL, ("C", "UTF-8"))
except locale.Error as e:
log.error("Could not switch to the C.UTF-8 locale: {}".format(e))
raise SystemExit
elif encoding != "UTF-8":
log.warning(
"Your locale {}.{} encoding is not UTF-8, switching to the UTF-8 version...".format(language, encoding))
try:
locale.setlocale(locale.LC_ALL, (language, "UTF-8"))
except locale.Error as e:
log.error("Could not set an UTF-8 encoding for the {} locale: {}".format(language, e))
raise SystemExit
else:
log.info("Current locale is {}.{}".format(language, encoding))
def _parse_arguments(self, argv):
"""
Parse command line arguments and override local configuration
:params args: Array of command line arguments
"""
parser = argparse.ArgumentParser(description="GNS3 server version {}".format(__version__))
parser.add_argument("-v", "--version", help="show the version", action="version", version=__version__)
parser.add_argument("--host", help="run on the given host/IP address")
parser.add_argument("--port", help="run on the given port", type=int)
parser.add_argument("--ssl", action="store_true", help="run in SSL mode")
parser.add_argument("--config", help="Configuration file")
parser.add_argument("--certfile", help="SSL cert file")
parser.add_argument("--certkey", help="SSL key file")
parser.add_argument("-L", "--local", action="store_true", help="local mode (allows some insecure operations)")
parser.add_argument("-A", "--allow", action="store_true",
help="allow remote connections to local console ports")
parser.add_argument("-q", "--quiet", default=False, action="store_true", help="do not show logs on stdout")
parser.add_argument("-d", "--debug", default=False, action="store_true", help="show debug logs")
parser.add_argument("--logfile", help="send output to logfile instead of console")
parser.add_argument("--logmaxsize", default=10000000, help="maximum logfile size in bytes (default is 10MB)")
parser.add_argument("--logbackupcount", default=10,
help="number of historical log files to keep (default is 10)")
parser.add_argument("--logcompression", default=False, action="store_true",
help="compress inactive (historical) logs")
parser.add_argument("--daemon", action="store_true", help="start as a daemon")
parser.add_argument("--pid", help="store process pid")
parser.add_argument("--profile", help="Settings profile (blank will use default settings files)")
args = parser.parse_args(argv)
level = logging.INFO
if args.debug:
level = logging.DEBUG
self._stream_handler = init_logger(level,
logfile=args.logfile,
max_bytes=int(args.logmaxsize),
backup_count=int(args.logbackupcount),
compression=args.logcompression,
quiet=args.quiet)
try:
if args.config:
Config.instance(files=[args.config], profile=args.profile)
else:
Config.instance(profile=args.profile)
config = Config.instance().settings
except ValidationError:
sys.exit(1)
defaults = {
"host": config.Server.host,
"port": config.Server.port,
"ssl": config.Server.enable_ssl,
"certfile": config.Server.certfile,
"certkey": config.Server.certkey,
"local": config.Server.local,
"allow": config.Server.allow_remote_console
}
parser.set_defaults(**defaults)
return parser.parse_args(argv)
@staticmethod
def _set_config_defaults_from_command_line(args):
config = Config.instance().settings
config.Server.local = args.local
config.Server.allow_remote_console = args.allow
config.Server.host = args.host
config.Server.port = args.port
config.Server.certfile = args.certfile
config.Server.certkey = args.certkey
config.Server.enable_ssl = args.ssl
async def reload_server(self):
"""
Reload the server.
"""
await Controller.instance().reload()
def _signal_handling(self):
def signal_handler(signame, *args):
try:
if signame == "SIGHUP":
log.info("Server has got signal {}, reloading...".format(signame))
asyncio.ensure_future(self.reload_server())
else:
log.info("Server has got signal {}, exiting...".format(signame))
os.kill(os.getpid(), signal.SIGTERM)
except asyncio.CancelledError:
pass
signals = [] # SIGINT and SIGTERM are already registered by uvicorn
if sys.platform.startswith("win"):
signals.extend(["SIGBREAK"])
else:
signals.extend(["SIGHUP", "SIGQUIT"])
for signal_name in signals:
callback = functools.partial(signal_handler, signal_name)
if sys.platform.startswith("win"):
# add_signal_handler() is not yet supported on Windows
signal.signal(getattr(signal, signal_name), callback)
else:
loop = asyncio.get_event_loop()
loop.add_signal_handler(getattr(signal, signal_name), callback)
@staticmethod
def _kill_ghosts(self):
"""
Kill process from previous GNS3 session
"""
detect_process = ["vpcs", "ubridge", "dynamips"]
for proc in psutil.process_iter():
try:
name = proc.name().lower().split(".")[0]
if name in detect_process:
proc.kill()
log.warning("Killed ghost process %s", name)
except (OSError, psutil.NoSuchProcess, psutil.AccessDenied):
pass
@staticmethod
def _pid_lock(self, path):
"""
Write the file in a file on the system.
Check if the process is not already running.
"""
if os.path.exists(path):
pid = None
try:
with open(path) as f:
try:
pid = int(f.read())
os.kill(pid, 0) # kill returns an error if the process is not running
except (OSError, SystemError, ValueError):
pid = None
except OSError as e:
log.critical("Can't open pid file %s: %s", pid, str(e))
sys.exit(1)
if pid:
log.critical("GNS3 is already running pid: %d", pid)
sys.exit(1)
try:
with open(path, 'w+') as f:
f.write(str(os.getpid()))
except OSError as e:
log.critical("Can't write pid file %s: %s", path, str(e))
sys.exit(1)
def run(self):
args = self._parse_arguments(sys.argv[1:])
if args.daemon and sys.platform.startswith("win"):
log.critical("Daemon is not supported on Windows")
sys.exit(1)
if args.pid:
self._pid_lock(args.pid)
self._kill_ghosts()
log.info("GNS3 server version {}".format(__version__))
current_year = datetime.date.today().year
log.info("Copyright (c) 2007-{} GNS3 Technologies Inc.".format(current_year))
for config_file in Config.instance().get_config_files():
log.info("Config file {} loaded".format(config_file))
self._set_config_defaults_from_command_line(args)
config = Config.instance().settings
if config.Server.local:
log.warning("Local mode is enabled. Beware, clients will have full control on your filesystem")
if config.Server.enable_http_auth:
log.info("HTTP authentication is enabled with username '{}'".format(config.Server.user))
# we only support Python 3 version >= 3.6
if sys.version_info < (3, 6, 0):
raise SystemExit("Python 3.6 or higher is required")
log.info("Running with Python {major}.{minor}.{micro} and has PID {pid}".format(major=sys.version_info[0],
minor=sys.version_info[1],
micro=sys.version_info[2],
pid=os.getpid()))
# check for the correct locale (UNIX/Linux only)
self._locale_check()
try:
os.getcwd()
except FileNotFoundError:
log.critical("The current working directory doesn't exist")
return
CrashReport.instance()
host = config.Server.host
port = config.Server.port
PortManager.instance().console_host = host
self._signal_handling()
try:
log.info("Starting server on {}:{}".format(host, port))
# only show uvicorn access logs in debug mode
access_log = False
if log.getEffectiveLevel() == logging.DEBUG:
access_log = True
if config.Server.enable_ssl:
if sys.platform.startswith("win"):
log.critical("SSL mode is not supported on Windows")
raise SystemExit
log.info("SSL is enabled")
config = uvicorn.Config(app,
host=host,
port=port,
access_log=access_log,
ssl_certfile=config.Server.certfile,
ssl_keyfile=config.Server.certkey,
lifespan="on")
# overwrite uvicorn loggers with our own logger
for uvicorn_logger_name in ("uvicorn", "uvicorn.error"):
uvicorn_logger = logging.getLogger(uvicorn_logger_name)
uvicorn_logger.handlers = [self._stream_handler]
uvicorn_logger.propagate = False
if access_log:
uvicorn_logger = logging.getLogger("uvicorn.access")
uvicorn_logger.handlers = [self._stream_handler]
uvicorn_logger.propagate = False
server = uvicorn.Server(config)
loop = asyncio.get_event_loop()
loop.run_until_complete(server.serve())
except OSError as e:
# This is to ignore OSError: [WinError 0] The operation completed successfully exception on Windows.
if not sys.platform.startswith("win") or not e.winerror == 0:
raise
except Exception as e:
log.critical(f"Critical error while running the server: {e}", exc_info=1)
CrashReport.instance().capture_exception()
return
finally:
if args.pid:
log.info("Remove PID file %s", args.pid)
try:
os.remove(args.pid)
except OSError as e:
log.critical("Can't remove pid file %s: %s", args.pid, str(e))

2
tests/conftest.py
View File

@ -353,7 +353,7 @@ def run_around_tests(monkeypatch, config, port_manager):#port_manager, controlle
config.settings.Server.appliances_path = os.path.join(tmppath, 'appliances')
config.settings.Server.ubridge_path = os.path.join(tmppath, 'bin', 'ubridge')
config.settings.Server.local = True
config.settings.Server.auth = False
config.settings.Server.enable_http_auth = False
# Prevent executions of the VM if we forgot to mock something
config.settings.VirtualBox.vboxmanage_path = tmppath

10
tests/test_config.py
View File

@ -121,9 +121,13 @@ def test_server_password_hidden():
({"vnc_console_end_port_range": 6000}, False),
({"vnc_console_end_port_range": 1000}, True),
({"vnc_console_start_port_range": 7000, "vnc_console_end_port_range": 6000}, True),
({"auth": True, "user": "user1"}, False),
({"auth": True, "user": ""}, True),
({"auth": True}, True),
({"enable_ssl": True, "certfile": "/path/to/certfile", "certkey": "/path/to/certkey"}, True),
({"enable_ssl": True}, True),
({"enable_ssl": True, "certfile": "/path/to/certfile"}, True),
({"enable_ssl": True, "certkey": "/path/to/certkey"}, True),
({"enable_http_auth": True, "user": "user1"}, False),
({"enable_http_auth": True, "user": ""}, True),
({"enable_http_auth": True}, True),
)
)
def test_server_settings(settings: dict, exception_expected: bool):

139
tests/test_run.py
View File

@ -1,139 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 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 pytest
import locale
from gns3server import run
from gns3server.config import Config
from gns3server.version import __version__
def test_locale_check():
try:
locale.setlocale(locale.LC_ALL, ("fr_FR", "UTF-8"))
except: # Locale is not available on the server
return
run.locale_check()
assert locale.getlocale() == ('fr_FR', 'UTF-8')
def test_parse_arguments(capsys, config, tmpdir):
server_config = config.settings.Server
with pytest.raises(SystemExit):
run.parse_arguments(["--fail"])
out, err = capsys.readouterr()
assert "usage" in err
assert "fail" in err
assert "unrecognized arguments" in err
# with pytest.raises(SystemExit):
# run.parse_arguments(["-v"])
# out, _ = capsys.readouterr()
# assert __version__ in out
# with pytest.raises(SystemExit):
# run.parse_arguments(["--version"])
# out, _ = capsys.readouterr()
# assert __version__ in out
#
# with pytest.raises(SystemExit):
# run.parse_arguments(["-h"])
# out, _ = capsys.readouterr()
# assert __version__ in out
# assert "optional arguments" in out
#
# with pytest.raises(SystemExit):
# run.parse_arguments(["--help"])
# out, _ = capsys.readouterr()
# assert __version__ in out
# assert "optional arguments" in out
assert run.parse_arguments(["--host", "192.168.1.1"]).host == "192.168.1.1"
assert run.parse_arguments([]).host == "0.0.0.0"
server_config.host = "192.168.1.2"
assert run.parse_arguments(["--host", "192.168.1.1"]).host == "192.168.1.1"
assert run.parse_arguments([]).host == "192.168.1.2"
assert run.parse_arguments(["--port", "8002"]).port == 8002
assert run.parse_arguments([]).port == 3080
server_config.port = 8003
assert run.parse_arguments([]).port == 8003
assert run.parse_arguments(["--ssl"]).ssl
assert run.parse_arguments([]).ssl is False
server_config.ssl = True
assert run.parse_arguments([]).ssl
assert run.parse_arguments(["--certfile", "bla"]).certfile == "bla"
assert run.parse_arguments([]).certfile is None
assert run.parse_arguments(["--certkey", "blu"]).certkey == "blu"
assert run.parse_arguments([]).certkey is None
assert run.parse_arguments(["-L"]).local
assert run.parse_arguments(["--local"]).local
server_config.local = False
assert run.parse_arguments([]).local is False
server_config.local = True
assert run.parse_arguments([]).local
assert run.parse_arguments(["-A"]).allow
assert run.parse_arguments(["--allow"]).allow
assert run.parse_arguments([]).allow is False
server_config.allow_remote_console = True
assert run.parse_arguments([]).allow
assert run.parse_arguments(["-q"]).quiet
assert run.parse_arguments(["--quiet"]).quiet
assert run.parse_arguments([]).quiet is False
assert run.parse_arguments(["-d"]).debug
assert run.parse_arguments([]).debug is False
server_config.debug = True
assert run.parse_arguments([]).debug
def test_set_config_with_args():
config = Config.instance()
args = run.parse_arguments(["--host",
"192.168.1.1",
"--local",
"--allow",
"--port",
"8001",
"--ssl",
"--certfile",
"bla",
"--certkey",
"blu",
"--debug"])
run.set_config(args)
server_config = config.settings.Server
assert server_config.local
assert server_config.allow_remote_console
assert server_config.host
assert server_config.port
assert server_config.ssl
assert server_config.certfile
assert server_config.certkey
assert server_config.debug

148
tests/test_server.py Normal file
View File

@ -0,0 +1,148 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 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 pytest
import locale
import tempfile
from gns3server.server import Server
from gns3server.config import Config
def test_locale_check():
try:
locale.setlocale(locale.LC_ALL, ("fr_FR", "UTF-8"))
except: # Locale is not available on the server
return
Server._locale_check()
assert locale.getlocale() == ('fr_FR', 'UTF-8')
def test_parse_arguments(capsys, config, tmpdir):
server = Server()
server_config = config.settings.Server
with pytest.raises(SystemExit):
server._parse_arguments(["--fail"])
out, err = capsys.readouterr()
assert "usage" in err
assert "fail" in err
assert "unrecognized arguments" in err
# with pytest.raises(SystemExit):
# run.parse_arguments(["-v"])
# out, _ = capsys.readouterr()
# assert __version__ in out
# with pytest.raises(SystemExit):
# run.parse_arguments(["--version"])
# out, _ = capsys.readouterr()
# assert __version__ in out
#
# with pytest.raises(SystemExit):
# run.parse_arguments(["-h"])
# out, _ = capsys.readouterr()
# assert __version__ in out
# assert "optional arguments" in out
#
# with pytest.raises(SystemExit):
# run.parse_arguments(["--help"])
# out, _ = capsys.readouterr()
# assert __version__ in out
# assert "optional arguments" in out
assert server._parse_arguments(["--host", "192.168.1.1"]).host == "192.168.1.1"
assert server._parse_arguments([]).host == "0.0.0.0"
server_config.host = "192.168.1.2"
assert server._parse_arguments(["--host", "192.168.1.1"]).host == "192.168.1.1"
assert server._parse_arguments([]).host == "192.168.1.2"
assert server._parse_arguments(["--port", "8002"]).port == 8002
assert server._parse_arguments([]).port == 3080
server_config.port = 8003
assert server._parse_arguments([]).port == 8003
assert server._parse_arguments(["--ssl"]).ssl
assert server._parse_arguments([]).ssl is False
with tempfile.NamedTemporaryFile(dir=str(tmpdir)) as f:
server_config.certfile = f.name
server_config.certkey = f.name
server_config.enable_ssl = True
assert server._parse_arguments([]).ssl
server_config.certfile = None
server_config.certkey = None
assert server._parse_arguments(["--certfile", "bla"]).certfile == "bla"
assert server._parse_arguments([]).certfile is None
assert server._parse_arguments(["--certkey", "blu"]).certkey == "blu"
assert server._parse_arguments([]).certkey is None
assert server._parse_arguments(["-L"]).local
assert server._parse_arguments(["--local"]).local
server_config.local = False
assert server._parse_arguments([]).local is False
server_config.local = True
assert server._parse_arguments([]).local
assert server._parse_arguments(["-A"]).allow
assert server._parse_arguments(["--allow"]).allow
assert server._parse_arguments([]).allow is False
server_config.allow_remote_console = True
assert server._parse_arguments([]).allow
assert server._parse_arguments(["-q"]).quiet
assert server._parse_arguments(["--quiet"]).quiet
assert server._parse_arguments([]).quiet is False
assert server._parse_arguments(["-d"]).debug
assert server._parse_arguments(["--debug"]).debug
assert server._parse_arguments([]).debug is False
def test_set_config_with_args(tmpdir):
server = Server()
config = Config.instance()
with tempfile.NamedTemporaryFile(dir=str(tmpdir)) as f:
certfile = f.name
certkey = f.name
args = server._parse_arguments(["--host",
"192.168.1.1",
"--local",
"--allow",
"--port",
"8001",
"--ssl",
"--certfile",
certfile,
"--certkey",
certkey,
"--debug"])
server._set_config_defaults_from_command_line(args)
server_config = config.settings.Server
assert server_config.local
assert server_config.allow_remote_console
assert server_config.host
assert server_config.port
assert server_config.enable_ssl
assert server_config.certfile
assert server_config.certkey