in gui/backend/gui_plugin/start/Start.py [0:0]
def web_server(port=None, secure=None, webrootpath=None,
single_instance_token=None, read_token_on_stdin=False,
accept_remote_connections=False, single_server=None):
"""Starts a web server that will serve the MySQL Shell GUI
Args:
port (int): The optional port the web server should be running on,
defaults to 8000
secure (dict): A dict with keyfile and certfile keys and values. An
empty dict will use the default key and certificate files. If
'None' is passed, then SSL will not be used.
webrootpath (str): The optional web root path that will be used
by the web server to serve files
single_instance_token (str): A token string used to establish
local user mode.
read_token_on_stdin (bool): If set to True, the token will be read
from STDIN
accept_remote_connections (bool): If set to True, the web server will
accept remote connections
single_server (str): The optional single server connection string
Allowed options for secure:
keyfile (str): The path to the server private key file
certfile (str): The path to the server certificate file
Returns:
Nothing
"""
import mysqlsh.plugin_manager.general
if (single_instance_token is not None or read_token_on_stdin) and single_server is not None:
raise ValueError(
'The single_instance_token and single_server parameters are mutually exclusive.')
# TODO: TEMPORARY HACK!!
# import gc
# gc.disable()
# Read single_instance_token from STDIN if read_token_on_stdin is True
if read_token_on_stdin:
single_instance_token = input(
'Please enter the single instance token: ').strip()
if not single_instance_token:
raise ValueError("No single instance token given on STDIN.")
logger.info('Token read from STDIN')
if platform.system() == 'Darwin':
result = subprocess.run(['ulimit', '-a'], stdout=subprocess.PIPE, check=False)
logger.debug(f"ULIMIT:\n{result.stdout.decode('utf-8')}")
# Start the web server
logger.info('Starting MySQL Shell GUI web server...')
server = None
try:
core_path = os.path.abspath(os.path.join(
os.path.dirname(__file__), '..', 'core'))
cert_path = mysqlsh.plugin_manager.general.get_shell_user_dir(
"plugin_data", "gui_plugin", "web_certs")
# Set defaults when necessary
if port is None:
port = 8000
if webrootpath is None:
webrootpath = os.path.join(core_path, 'webroot')
# cspell:ignore chdir
if not os.path.isdir(webrootpath):
raise Exception(
'Cannot start webserver. Root directory does not '
f'exist({webrootpath}).')
os.chdir(webrootpath)
# Check if we can supply a default page
if not os.path.isfile("index.html"):
raise Exception(
'Cannot start webserver. The "index.html" file does not '
f'exist in {webrootpath}.')
if secure is not None:
# try to cast from shell.Dict to dict
try:
secure = json.loads(str(secure))
except:
pass
default_cert_used = False
if type(secure) is dict:
if 'keyfile' not in secure or secure['keyfile'] == "default":
default_cert_used = True
secure['keyfile'] = path.join(*[
cert_path,
'server.key'])
if 'certfile' not in secure or secure['certfile'] == "default":
default_cert_used = True
secure['certfile'] = path.join(*[
cert_path,
'server.crt'])
else:
raise ValueError('If specified, the secure parameter need to '
'be of type dict')
# If the default cert is used, check if it is already installed
if not accept_remote_connections:
logger.info('\tChecking web server certificate...')
if default_cert_used:
try:
if not is_shell_web_certificate_installed(check_keychain=True):
logger.info('\tCertificate is not installed. '
'Use gui.core.installShellWebCertificate() to install one.')
return
except Exception as e:
logger.info('\tCertificate is not correctly installed. '
'Use gui.core.installShellWebCertificate() to fix the installation.')
logger.exception(e)
return
logger.info('\tCertificate is installed.')
# Replace WSSimpleEcho with your own subclass of HTTPWebSocketHandler
server = ThreadedHTTPServer(
('127.0.0.1' if not accept_remote_connections else '0.0.0.0', port), ShellGuiWebSocketHandler)
server.daemon_threads = True
server.host = f'{"https" if secure else "http"}://{"127.0.0.1" if not accept_remote_connections else socket.getfqdn()}'
server.port = port
server.single_instance_token = single_instance_token
server.single_server = single_server
if secure:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(
certfile=secure['certfile'], keyfile=secure['keyfile'])
server.socket = context.wrap_socket(
server.socket,
server_side=True)
def user_signal_handler(signum, frame):
server.force_stop()
# Register signal handler for SIGINT (handle ctrl+c)
signal.signal(signal.SIGINT, user_signal_handler)
try:
logger.info(
f"Server started [port:{port}, "
f"secure:{'version' in dir(server.socket)}, "
f"single user: {server.single_instance_token is not None}]"
f"single server: {server.single_server is not None}]",
['session'])
# Log server start
logger.info(f"\tPort: {port}")
logger.info(f"\tSecure: {'version' in dir(server.socket)}")
logger.info(f"\tWebroot: {webrootpath}")
logger.info(f"\tAccept remote connections: {accept_remote_connections}")
if server.single_instance_token is not None:
mode = 'Single user'
elif server.single_server is not None:
mode = 'Single server'
else:
mode = 'Multi-user'
logger.info(f"\tMode: {mode}")
if server.single_instance_token:
logger.add_filter({
"type": "substring",
"start": "token=",
"end": " HTTP",
"expire": Filtering.FilterExpire.OnUse if SystemUtils.in_vs_code() else Filtering.FilterExpire.Never
})
# Start web server
server.serve_forever()
# TODO(anyone): Using the 'session' tag here causes database locks
# logger.info("Web server is down.", ['session'])
logger.info("Web server is down.")
except Exception as e: # pragma: no cover
logger.error(f'Log message could not be inserted into db. {e}')
except KeyboardInterrupt: # pragma: no cover
logger.info('^C received, shutting down server')
finally:
if server:
server.socket.close()