notebook/notebook/notebookapp.py at v6.5.2 · jupyter/notebook

History
"""A tornado based Jupyter notebook server."""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
import notebook
import asyncio
import binascii
import datetime
import errno
import functools
import gettext
import hashlib
import hmac
import importlib
import inspect
import ipaddress
import json
import logging
import mimetypes
import random
import select
import signal
import socket
import stat
import tempfile
import threading
import time
import warnings
import webbrowser
    import resource
except ImportError:
    # Windows
    resource = None
from base64 import encodebytes
from jinja2 import Environment, FileSystemLoader
from notebook.transutils import trans, _
# check for tornado 3.1.0
    import tornado
except ImportError as e:
    raise ImportError(_("The Jupyter Notebook requires tornado >= 5.0")) from e
    version_info = tornado.version_info
except AttributeError as e:
    raise ImportError(_("The Jupyter Notebook requires tornado >= 5.0, but you have < 1.1.0")) from e
if version_info < (5,0):
    raise ImportError(_("The Jupyter Notebook requires tornado >= 5.0, but you have %s") % tornado.version)
from tornado import httpserver
from tornado import ioloop
from tornado import web
from tornado.httputil import url_concat
from tornado.log import LogFormatter, app_log, access_log, gen_log
if not sys.platform.startswith('win'):
    from tornado.netutil import bind_unix_socket
from notebook import (
    DEFAULT_NOTEBOOK_PORT,
    DEFAULT_TEMPLATE_PATH_LIST,
    __version__,
import nbclassic
# Packagers: modify this line if you store the notebook static files elsewhere
DEFAULT_STATIC_FILES_PATH = os.path.join(os.path.dirname(nbclassic.__file__), "static")
from .base.handlers import Template404, RedirectWithParams
from .log import log_request
from .services.kernels.kernelmanager import MappingKernelManager, AsyncMappingKernelManager
from .services.config import ConfigManager
from .services.contents.manager import ContentsManager
from .services.contents.filemanager import FileContentsManager
from .services.contents.largefilemanager import LargeFileManager
from .services.sessions.sessionmanager import SessionManager
from .gateway.managers import GatewayKernelManager, GatewayKernelSpecManager, GatewaySessionManager, GatewayClient
from .auth.login import LoginHandler
from .auth.logout import LogoutHandler
from .base.handlers import FileFindHandler
from traitlets.config import Config
from traitlets.config.application import catch_config_error, boolean_flag
from jupyter_core.application import (
    JupyterApp, base_flags, base_aliases,
from jupyter_core.paths import jupyter_config_path
from jupyter_client import KernelManager
from jupyter_client.kernelspec import KernelSpecManager
from jupyter_client.session import Session
from nbformat.sign import NotebookNotary
from traitlets import (
    Any, Dict, Unicode, Integer, List, Bool, Bytes, Instance,
    TraitError, Type, Float, observe, default, validate
from ipython_genutils import py3compat
from jupyter_core.paths import jupyter_runtime_dir, jupyter_path
from notebook._sysinfo import get_sys_info
from ._tz import utcnow, utcfromtimestamp
from .utils import (
    check_pid,
    pathname2url,
    run_sync,
    unix_socket_in_use,
    url_escape,
    url_path_join,
    urldecode_unix_socket_path,
    urlencode_unix_socket,
    urljoin,
from .traittypes import TypeFromClasses
# Check if we can use async kernel management
    from jupyter_client import AsyncMultiKernelManager
    async_kernel_mgmt_available = True
except ImportError:
    async_kernel_mgmt_available = False
# Tolerate missing terminado package.
    from .terminal import TerminalManager
    terminado_available = True
except ImportError:
    terminado_available = False
#-----------------------------------------------------------------------------
# Module globals
#-----------------------------------------------------------------------------
_examples = """
jupyter notebook                       # start the notebook
jupyter notebook --certfile=mycert.pem # use SSL/TLS certificate
jupyter notebook password              # enter a password to protect the server
#-----------------------------------------------------------------------------
# Helper functions
#-----------------------------------------------------------------------------
def random_ports(port, n):
    """Generate a list of n random ports near the given port.
    The first 5 ports will be sequential, and the remaining n-5 will be
    randomly selected in the range [port-2*n, port+2*n].
    for i in range(min(5, n)):
        yield port + i
    for i in range(n-5):
        yield max(1, port + random.randint(-2*n, 2*n))
def load_handlers(name):
    """Load the (URL pattern, handler) tuples for each component."""
    mod = __import__(name, fromlist=['default_handlers'])
    return mod.default_handlers
#-----------------------------------------------------------------------------
# The Tornado web application
#-----------------------------------------------------------------------------
class NotebookWebApplication(web.Application):
    def __init__(self, jupyter_app, kernel_manager, contents_manager,
                 session_manager, kernel_spec_manager,
                 config_manager, extra_services, log,
                 base_url, default_url, settings_overrides, jinja_env_options):
        settings = self.init_settings(
            jupyter_app, kernel_manager, contents_manager,
            session_manager, kernel_spec_manager, config_manager,
            extra_services, log, base_url,
            default_url, settings_overrides, jinja_env_options)
        handlers = self.init_handlers(settings)
        if settings['autoreload']:
            log.info('Autoreload enabled: the webapp will restart when any Python src file changes.')
        super().__init__(handlers, **settings)
    def init_settings(self, jupyter_app, kernel_manager, contents_manager,
                      session_manager, kernel_spec_manager,
                      config_manager, extra_services,
                      log, base_url, default_url, settings_overrides,
                      jinja_env_options=None):
        _template_path = settings_overrides.get(
            "template_path",
            jupyter_app.template_file_path,
        if isinstance(_template_path, py3compat.string_types):
            _template_path = (_template_path,)
        template_path = [os.path.expanduser(path) for path in _template_path]
        jenv_opt = {"autoescape": True}
        jenv_opt.update(jinja_env_options if jinja_env_options else {})
        env = Environment(loader=FileSystemLoader(template_path), extensions=['jinja2.ext.i18n'], **jenv_opt)
        sys_info = get_sys_info()
        # If the user is running the notebook in a git directory, make the assumption
        # that this is a dev install and suggest to the developer `npm run build:watch`.
        base_dir = os.path.realpath(os.path.join(__file__, '..', '..'))
        dev_mode = os.path.exists(os.path.join(base_dir, '.git'))
        nbui = gettext.translation('nbui', localedir=os.path.join(base_dir, 'notebook/i18n'), fallback=True)
        env.install_gettext_translations(nbui, newstyle=False)
        if dev_mode:
            DEV_NOTE_NPM = f"""It looks like you're running the notebook from source.
    If you're working on the Javascript of the notebook, try running
    {'npm run build:watch'}
    in another terminal window to have the system incrementally
    watch and build the notebook's JavaScript for you, as you make changes."""
            log.info(DEV_NOTE_NPM)
        if sys_info['commit_source'] == 'repository':
            # don't cache (rely on 304) when working from master
            version_hash = ''
        else:
            # reset the cache on server restart
            version_hash = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
        if jupyter_app.ignore_minified_js:
            log.warning(_("""The `ignore_minified_js` flag is deprecated and no longer works."""))
            log.warning(_("""Alternatively use `%s` when working on the notebook's Javascript and LESS""") % 'npm run build:watch')
            warnings.warn(_("The `ignore_minified_js` flag is deprecated and will be removed in Notebook 6.0"), DeprecationWarning)
        now = utcnow()
        root_dir = contents_manager.root_dir
        home = py3compat.str_to_unicode(os.path.expanduser('~'), encoding=sys.getfilesystemencoding())
        if root_dir.startswith(home + os.path.sep):
            # collapse $HOME to ~
            root_dir = '~' + root_dir[len(home):]
        # Use the NotebookApp logger and its formatting for tornado request logging.
        log_function = functools.partial(
            log_request, log=log, log_json=jupyter_app.log_json)
        settings = dict(
            # basics
            log_function=log_function,
            base_url=base_url,
            default_url=default_url,
            template_path=template_path,
            static_path=jupyter_app.static_file_path,
            static_custom_path=jupyter_app.static_custom_path,
            static_handler_class = FileFindHandler,
            static_url_prefix = url_path_join(base_url,'/static/'),
            static_handler_args = {
                # don't cache custom.js
                'no_cache_paths': [url_path_join(base_url, 'static', 'custom')],
            version_hash=version_hash,
            ignore_minified_js=jupyter_app.ignore_minified_js,
            # rate limits
            iopub_msg_rate_limit=jupyter_app.iopub_msg_rate_limit,
            iopub_data_rate_limit=jupyter_app.iopub_data_rate_limit,
            rate_limit_window=jupyter_app.rate_limit_window,
            # authentication
            cookie_secret=jupyter_app.cookie_secret,
            login_url=url_path_join(base_url,'/login'),
            login_handler_class=jupyter_app.login_handler_class,
            logout_handler_class=jupyter_app.logout_handler_class,
            password=jupyter_app.password,
            xsrf_cookies=True,
            disable_check_xsrf=jupyter_app.disable_check_xsrf,
            allow_remote_access=jupyter_app.allow_remote_access,
            local_hostnames=jupyter_app.local_hostnames,
            authenticate_prometheus=jupyter_app.authenticate_prometheus,
            # managers
            kernel_manager=kernel_manager,
            contents_manager=contents_manager,
            session_manager=session_manager,
            kernel_spec_manager=kernel_spec_manager,
            config_manager=config_manager,
            # handlers
            extra_services=extra_services,
            # Jupyter stuff
            started=now,
            # place for extensions to register activity
            # so that they can prevent idle-shutdown
            last_activity_times={},
            jinja_template_vars=jupyter_app.jinja_template_vars,
            nbextensions_path=jupyter_app.nbextensions_path,
            websocket_url=jupyter_app.websocket_url,
            mathjax_url=jupyter_app.mathjax_url,
            mathjax_config=jupyter_app.mathjax_config,
            shutdown_button=jupyter_app.quit_button,
            config=jupyter_app.config,
            config_dir=jupyter_app.config_dir,
            allow_password_change=jupyter_app.allow_password_change,
            server_root_dir=root_dir,
            jinja2_env=env,
            terminals_available=terminado_available and jupyter_app.terminals_enabled,
        # allow custom overrides for the tornado web app.
        settings.update(settings_overrides)
        return settings
    def init_handlers(self, settings):
        """Load the (URL pattern, handler) tuples for each component."""
        # Order matters. The first handler to match the URL will handle the request.
        handlers = []
        # load extra services specified by users before default handlers
        for service in settings['extra_services']:
            handlers.extend(load_handlers(service))
        handlers.extend(load_handlers('notebook.tree.handlers'))
        handlers.extend([(r"/login", settings['login_handler_class'])])
        handlers.extend([(r"/logout", settings['logout_handler_class'])])
        handlers.extend(load_handlers('notebook.files.handlers'))
        handlers.extend(load_handlers('notebook.view.handlers'))
        handlers.extend(load_handlers('notebook.notebook.handlers'))
        handlers.extend(load_handlers('notebook.nbconvert.handlers'))
        handlers.extend(load_handlers('notebook.bundler.handlers'))
        handlers.extend(load_handlers('notebook.kernelspecs.handlers'))
        handlers.extend(load_handlers('notebook.edit.handlers'))
        handlers.extend(load_handlers('notebook.services.api.handlers'))
        handlers.extend(load_handlers('notebook.services.config.handlers'))
        handlers.extend(load_handlers('notebook.services.contents.handlers'))
        handlers.extend(load_handlers('notebook.services.sessions.handlers'))
        handlers.extend(load_handlers('notebook.services.nbconvert.handlers'))
        handlers.extend(load_handlers('notebook.services.security.handlers'))
        handlers.extend(load_handlers('notebook.services.shutdown'))
        handlers.extend(load_handlers('notebook.services.kernels.handlers'))
        handlers.extend(load_handlers('notebook.services.kernelspecs.handlers'))
        handlers.extend(settings['contents_manager'].get_extra_handlers())
        # If gateway mode is enabled, replace appropriate handlers to perform redirection
        if GatewayClient.instance().gateway_enabled:
            # for each handler required for gateway, locate its pattern
            # in the current list and replace that entry...
            gateway_handlers = load_handlers('notebook.gateway.handlers')
            for i, gwh in enumerate(gateway_handlers):
                for j, h in enumerate(handlers):
                    if gwh[0] == h[0]:
                        handlers[j] = (gwh[0], gwh[1])
        handlers.append(
            (r"/nbextensions/(.*)", FileFindHandler, {
                'path': settings['nbextensions_path'],
                'no_cache_paths': ['/'], # don't cache anything in nbextensions
        handlers.append(
            (r"/custom/(.*)", FileFindHandler, {
                'path': settings['static_custom_path'],
                'no_cache_paths': ['/'], # don't cache anything in custom
        # register base handlers last
        handlers.extend(load_handlers('notebook.base.handlers'))
        # set the URL that will be redirected from `/`
        handlers.append(
            (r'/?', RedirectWithParams, {
                'url' : settings['default_url'],
                'permanent': False, # want 302, not 301
        # prepend base_url onto the patterns that we match
        new_handlers = []
        for handler in handlers:
            pattern = url_path_join(settings['base_url'], handler[0])
            new_handler = tuple([pattern] + list(handler[1:]))
            new_handlers.append(new_handler)
        # add 404 on the end, which will catch everything that falls through
        new_handlers.append((r'(.*)', Template404))
        return new_handlers
    def last_activity(self):
        """Get a UTC timestamp for when the server last did something.
        Includes: API activity, kernel activity, kernel shutdown, and terminal
        activity.
        """
        sources = [
            self.settings['started'],
            self.settings['kernel_manager'].last_kernel_activity,
        try:
            sources.append(self.settings['api_last_activity'])
        except KeyError:
            pass
        try:
            sources.append(self.settings['terminal_last_activity'])
        except KeyError:
            pass
        sources.extend(self.settings['last_activity_times'].values())
        return max(sources)
class NotebookPasswordApp(JupyterApp):
    """Set a password for the notebook server.
    Setting a password secures the notebook server
    and removes the need for token-based authentication.
    description = __doc__
    def _config_file_default(self):
        return os.path.join(self.config_dir, 'jupyter_notebook_config.json')
    def start(self):
        from .auth.security import set_password
        set_password(config_file=self.config_file)
        self.log.info(f"Wrote hashed password to {self.config_file}")
def shutdown_server(server_info, timeout=5, log=None):
    """Shutdown a notebook server in a separate process.
    *server_info* should be a dictionary as produced by list_running_servers().
    Will first try to request shutdown using /api/shutdown .
    On Unix, if the server is still running after *timeout* seconds, it will
    send SIGTERM. After another timeout, it escalates to SIGKILL.
    Returns True if the server was stopped by any means, False if stopping it
    failed (on Windows).
    from tornado import gen
    from tornado.httpclient import AsyncHTTPClient, HTTPClient, HTTPRequest
    from tornado.netutil import Resolver
    url = server_info['url']
    pid = server_info['pid']
    resolver = None
    # UNIX Socket handling.
    if url.startswith('http+unix://'):
        # This library doesn't understand our URI form, but it's just HTTP.
        url = url.replace('http+unix://', 'http://')
        class UnixSocketResolver(Resolver):
            def initialize(self, resolver):
                self.resolver = resolver
            def close(self):
                self.resolver.close()
            @gen.coroutine
            def resolve(self, host, port, *args, **kwargs):
                raise gen.Return([
                    (socket.AF_UNIX, urldecode_unix_socket_path(host))
        resolver = UnixSocketResolver(resolver=Resolver())
    req = HTTPRequest(url + 'api/shutdown', method='POST', body=b'', headers={
        'Authorization': 'token ' + server_info['token']
    if log: log.debug("POST request to %sapi/shutdown", url)
    AsyncHTTPClient.configure(None, resolver=resolver)
    HTTPClient(AsyncHTTPClient).fetch(req)
    # Poll to see if it shut down.
    for _ in range(timeout*10):
        if not check_pid(pid):
            if log: log.debug("Server PID %s is gone", pid)
            return True
        time.sleep(0.1)
    if sys.platform.startswith('win'):
        return False
    if log: log.debug("SIGTERM to PID %s", pid)
    os.kill(pid, signal.SIGTERM)
    # Poll to see if it shut down.
    for _ in range(timeout * 10):
        if not check_pid(pid):
            if log: log.debug("Server PID %s is gone", pid)
            return True
        time.sleep(0.1)
    if log: log.debug("SIGKILL to PID %s", pid)
    os.kill(pid, signal.SIGKILL)
    return True  # SIGKILL cannot be caught
class NbserverStopApp(JupyterApp):
    version = __version__
    description="Stop currently running notebook server."
    port = Integer(DEFAULT_NOTEBOOK_PORT, config=True,
        help=f"Port of the server to be killed. Default {DEFAULT_NOTEBOOK_PORT}")
    sock = Unicode('', config=True,
        help="UNIX socket of the server to be killed.")
    def parse_command_line(self, argv=None):
        super().parse_command_line(argv)
        if self.extra_args:
            try:
                self.port = int(self.extra_args[0])
            except ValueError:
                # self.extra_args[0] was not an int, so it must be a string (unix socket).
                self.sock = self.extra_args[0]
    def shutdown_server(self, server):
        return shutdown_server(server, log=self.log)
    def _shutdown_or_exit(self, target_endpoint, server):
        print(f"Shutting down server on {target_endpoint}...")
        server_stopped = self.shutdown_server(server)
        if not server_stopped and sys.platform.startswith('win'):
            # the pid check on Windows appears to be unreliable, so fetch another
            # list of servers and ensure our server is not in the list before
            # sending the wrong impression.
            servers = list(list_running_servers(self.runtime_dir))
            if server not in servers:
                server_stopped = True
        if not server_stopped:
            sys.exit(f"Could not stop server on {target_endpoint}")
    @staticmethod
    def _maybe_remove_unix_socket(socket_path):
        try:
            os.unlink(socket_path)
        except OSError:
            pass
    def start(self):
        servers = list(list_running_servers(self.runtime_dir))
        if not servers:
            self.exit("There are no running servers (per %s)" % self.runtime_dir)
        for server in servers:
            if self.sock:
                sock = server.get('sock', None)
                if sock and sock == self.sock:
                    self._shutdown_or_exit(sock, server)
                    # Attempt to remove the UNIX socket after stopping.
                    self._maybe_remove_unix_socket(sock)
                    return
            elif self.port:
                port = server.get('port', None)
                if port == self.port:
                    self._shutdown_or_exit(port, server)
                    return
        else:
            current_endpoint = self.sock or self.port
            print(
                f"There is currently no server running on {current_endpoint}",
                file=sys.stderr
            print("Ports/sockets currently in use:", file=sys.stderr)
            for server in servers:
                print("  - {}".format(server.get('sock') or server['port']), file=sys.stderr)
            self.exit(1)
class NbserverListApp(JupyterApp):
    version = __version__
    description=_("List currently running notebook servers.")
    flags = dict(
        jsonlist=({'NbserverListApp': {'jsonlist': True}},
              _("Produce machine-readable JSON list output.")),
        json=({'NbserverListApp': {'json': True}},
              _("Produce machine-readable JSON object on each line of output.")),
    jsonlist = Bool(False, config=True,
          help=_("If True, the output will be a JSON list of objects, one per "
                 "active notebook server, each with the details from the "
                 "relevant server info file."))
    json = Bool(False, config=True,
          help=_("If True, each line of output will be a JSON object with the "
                  "details from the server info file. For a JSON list output, "
                  "see the NbserverListApp.jsonlist configuration value"))
    def start(self):
        serverinfo_list = list(list_running_servers(self.runtime_dir))
        if self.jsonlist:
            print(json.dumps(serverinfo_list, indent=2))
        elif self.json:
            for serverinfo in serverinfo_list:
                print(json.dumps(serverinfo))
        else:
            print("Currently running servers:")
            for serverinfo in serverinfo_list:
                url = serverinfo['url']
                if serverinfo.get('token'):
                    url = url + '?token=%s' % serverinfo['token']
                print(url, "::", serverinfo['notebook_dir'])
#-----------------------------------------------------------------------------
# Aliases and Flags
#-----------------------------------------------------------------------------
flags = dict(base_flags)
flags['no-browser']=(
    {'NotebookApp' : {'open_browser' : False}},
    _("Don't open the notebook in a browser after startup.")
flags['pylab']=(
    {'NotebookApp' : {'pylab' : 'warn'}},
    _("DISABLED: use %pylab or %matplotlib in the notebook to enable matplotlib.")
flags['no-mathjax']=(
    {'NotebookApp' : {'enable_mathjax' : False}},
    """Disable MathJax
    MathJax is the javascript library Jupyter uses to render math/LaTeX. It is
    very large, so you may want to disable it if you have a slow internet
    connection, or for offline use of the notebook.
    When disabled, equations etc. will appear as their untransformed TeX source.
flags['allow-root']=(
    {'NotebookApp' : {'allow_root' : True}},
    _("Allow the notebook to be run from root user.")
flags['autoreload'] = (
    {'NotebookApp': {'autoreload': True}},
    """Autoreload the webapp
    Enable reloading of the tornado webapp and all imported Python packages
    when any changes are made to any Python src files in Notebook or
    extensions.
# Add notebook manager flags
flags.update(boolean_flag('script', 'FileContentsManager.save_script',
               'DEPRECATED, IGNORED',
               'DEPRECATED, IGNORED'))
aliases = dict(base_aliases)
aliases.update({
    'ip': 'NotebookApp.ip',
    'port': 'NotebookApp.port',
    'port-retries': 'NotebookApp.port_retries',
    'sock': 'NotebookApp.sock',
    'sock-mode': 'NotebookApp.sock_mode',
    'transport': 'KernelManager.transport',
    'keyfile': 'NotebookApp.keyfile',
    'certfile': 'NotebookApp.certfile',
    'client-ca': 'NotebookApp.client_ca',
    'notebook-dir': 'NotebookApp.notebook_dir',
    'browser': 'NotebookApp.browser',
    'pylab': 'NotebookApp.pylab',
    'gateway-url': 'GatewayClient.url',
#-----------------------------------------------------------------------------
# NotebookApp
#-----------------------------------------------------------------------------
class NotebookApp(JupyterApp):
    name = 'jupyter-notebook'
    version = __version__
    description = _("""The Jupyter HTML Notebook.
    This launches a Tornado based HTML Notebook Server that serves up an HTML5/Javascript Notebook client.""")
    examples = _examples
    aliases = aliases
    flags = flags
    classes = [
        KernelManager, Session, MappingKernelManager, KernelSpecManager,
        ContentsManager, FileContentsManager, NotebookNotary,
        GatewayKernelManager, GatewayKernelSpecManager, GatewaySessionManager, GatewayClient,
    if terminado_available:  # Only necessary when terminado is available
        classes.append(TerminalManager)
    flags = Dict(flags)
    aliases = Dict(aliases)
    subcommands = dict(
        list=(NbserverListApp, NbserverListApp.description.splitlines()[0]),
        stop=(NbserverStopApp, NbserverStopApp.description.splitlines()[0]),
        password=(NotebookPasswordApp, NotebookPasswordApp.description.splitlines()[0]),
    _log_formatter_cls = LogFormatter
    _json_logging_import_error_logged = False
    log_json = Bool(False, config=True,
        help=_('Set to True to enable JSON formatted logs. '
               'Run "pip install notebook[json-logging]" to install the '
               'required dependent packages. Can also be set using the '
               'environment variable JUPYTER_ENABLE_JSON_LOGGING=true.')
    @default('log_json')
    def _default_log_json(self):
        """Get the log_json value from the environment."""
        return os.getenv('JUPYTER_ENABLE_JSON_LOGGING', 'false').lower() == 'true'
    @validate('log_json')
    def _validate_log_json(self, proposal):
        # If log_json=True, see if the json_logging package can be imported and
        # override _log_formatter_cls if so.
        value = proposal['value']
        if value:
            try:
                import json_logging
                self.log.debug('initializing json logging')
                json_logging.init_non_web(enable_json=True)
                self._log_formatter_cls = json_logging.JSONLogFormatter
            except ImportError:
                # If configured for json logs and we can't do it, log a hint.
                # Only log the error once though.
                if not self._json_logging_import_error_logged:
                    self.log.warning(
                        'Unable to use json logging due to missing packages. '
                        'Run "pip install notebook[json-logging]" to fix.'
                    self._json_logging_import_error_logged = True
                value = False
        return value
    @default('log_level')
    def _default_log_level(self):
        return logging.INFO
    @default('log_datefmt')
    def _default_log_datefmt(self):
        """Exclude date from default date format"""
        return "%H:%M:%S"
    @default('log_format')
    def _default_log_format(self):
        """override default log format to include time"""
        return "%(color)s[%(levelname)1.1s %(asctime)s.%(msecs).03d %(name)s]%(end_color)s %(message)s"
    ignore_minified_js = Bool(False,
            config=True,
            help=_('Deprecated: Use minified JS file or not, mainly use during dev to avoid JS recompilation'),
    # file to be opened in the notebook server
    file_to_run = Unicode('', config=True)
    # Network related information
    allow_origin = Unicode('', config=True,
        help="""Set the Access-Control-Allow-Origin header
        Use '*' to allow any origin to access your server.
        Takes precedence over allow_origin_pat.
        """
    allow_origin_pat = Unicode('', config=True,
        help="""Use a regular expression for the Access-Control-Allow-Origin header
        Requests from an origin matching the expression will get replies with:
            Access-Control-Allow-Origin: origin
        where `origin` is the origin of the request.
        Ignored if allow_origin is set.
        """
    allow_credentials = Bool(False, config=True,
        help=_("Set the Access-Control-Allow-Credentials: true header")
    allow_root = Bool(False, config=True,
        help=_("Whether to allow the user to run the notebook as root.")
    use_redirect_file = Bool(True, config=True,
        help="""Disable launching browser by redirect file
        For versions of notebook > 5.7.2, a security feature measure was added that
        prevented the authentication token used to launch the browser from being visible.
        This feature makes it difficult for other users on a multi-user system from
        running code in your Jupyter session as you.
        However, some environments (like Windows Subsystem for Linux (WSL) and Chromebooks),
        launching a browser using a redirect file can lead the browser failing to load.
        This is because of the difference in file structures/paths between the runtime and
        the browser.
        Disabling this setting to False will disable this behavior, allowing the browser
        to launch by using a URL and visible token (as before).
        """
    autoreload = Bool(False, config=True,
        help= ("Reload the webapp when changes are made to any Python src files.")
    default_url = Unicode('/tree', config=True,
        help=_("The default URL to redirect to from `/`")
    ip = Unicode('localhost', config=True,
        help=_("The IP address the notebook server will listen on.")
    @default('ip')
    def _default_ip(self):
        """Return localhost if available, 127.0.0.1 otherwise.
        On some (horribly broken) systems, localhost cannot be bound.
        """
        s = socket.socket()
        try:
            s.bind(('localhost', 0))
        except OSError as e:
            self.log.warning(_("Cannot bind to localhost, using 127.0.0.1 as default ip\n%s"), e)
            return '127.0.0.1'
        else:
            s.close()
            return 'localhost'
    @validate('ip')
    def _validate_ip(self, proposal):
        value = proposal['value']
        if value == '*':
            value = ''
        return value
    custom_display_url = Unicode('', config=True,
        help=_("""Override URL shown to users.
        Replace actual URL, including protocol, address, port and base URL,
        with the given value when displaying URL to the users. Do not change
        the actual connection URL. If authentication token is enabled, the
        token is added to the custom URL automatically.
        This option is intended to be used when the URL to display to the user
        cannot be determined reliably by the Jupyter notebook server (proxified
        or containerized setups for example).""")
    port_env = 'JUPYTER_PORT'
    port_default_value = DEFAULT_NOTEBOOK_PORT
    port = Integer(port_default_value, config=True,
        help=_("The port the notebook server will listen on (env: JUPYTER_PORT).")
    @default('port')
    def port_default(self):
        return int(os.getenv(self.port_env, self.port_default_value))
    port_retries_env = 'JUPYTER_PORT_RETRIES'
    port_retries_default_value = 50
    port_retries = Integer(port_retries_default_value, config=True,
        help=_("The number of additional ports to try if the specified port is not "
               "available (env: JUPYTER_PORT_RETRIES).")
    @default('port_retries')
    def port_retries_default(self):
        return int(os.getenv(self.port_retries_env, self.port_retries_default_value))
    sock = Unicode('', config=True,
        help=_("The UNIX socket the notebook server will listen on.")
    sock_mode = Unicode('0600', config=True,
        help=_("The permissions mode for UNIX socket creation (default: 0600).")
    @validate('sock_mode')
    def _validate_sock_mode(self, proposal):
        value = proposal['value']
        try:
            converted_value = int(value.encode(), 8)
            assert all((
                # Ensure the mode is at least user readable/writable.
                bool(converted_value & stat.S_IRUSR),
                bool(converted_value & stat.S_IWUSR),
                # And isn't out of bounds.
                converted_value <= 2 ** 12
        except ValueError as e:
            raise TraitError(
                'invalid --sock-mode value: %s, please specify as e.g. "0600"' % value
            ) from e
        except AssertionError as e:
            raise TraitError(
                'invalid --sock-mode value: %s, must have u+rw (0600) at a minimum' % value
            ) from e
        return value
    certfile = Unicode('', config=True,
        help=_("""The full path to an SSL/TLS certificate file.""")
    keyfile = Unicode('', config=True,
       
87F8
 help=_("""The full path to a private key file for usage with SSL/TLS.""")
    client_ca = Unicode('', config=True,
        help=_("""The full path to a certificate authority certificate for SSL/TLS client authentication.""")
    cookie_secret_file = Unicode(config=True,
        help=_("""The file where the cookie secret is stored.""")
    @default('cookie_secret_file')
    def _default_cookie_secret_file(self):
        return os.path.join(self.runtime_dir, 'notebook_cookie_secret')
    cookie_secret = Bytes(b'', config=True,
        help="""The random bytes used to secure cookies.
        By default this is a new random number every time you start the Notebook.
        Set it to a value in a config file to enable logins to persist across server sessions.
        Note: Cookie secrets should be kept private, do not share config files with
        cookie_secret stored in plaintext (you can read the value from a file).
        """
    @default('cookie_secret')
    def _default_cookie_secret(self):
        if os.path.exists(self.cookie_secret_file):
            with open(self.cookie_secret_file, 'rb') as f:
                key =  f.read()
        else:
            key = encodebytes(os.urandom(32))
            self._write_cookie_secret_file(key)
        h = hmac.new(key, digestmod=hashlib.sha256)
        h.update(self.password.encode())
        return h.digest()
    def _write_cookie_secret_file(self, secret):
        """write my secret to my secret_file"""
        self.log.info(_("Writing notebook server cookie secret to %s"), self.cookie_secret_file)
        try:
            with open(self.cookie_secret_file, 'wb') as f:
                f.write(secret)
        except OSError as e:
            self.log.error(_("Failed to write cookie secret to %s: %s"),
                           self.cookie_secret_file, e)
        try:
            os.chmod(self.cookie_secret_file, 0o600)
        except OSError:
            self.log.warning(
                _("Could not set permissions on %s"),
                self.cookie_secret_file
    token = Unicode('<generated>',
        help=_("""Token used for authenticating first-time connections to the server.
        The token can be read from the file referenced by JUPYTER_TOKEN_FILE or set directly
        with the JUPYTER_TOKEN environment variable.
        When no password is enabled,
        the default is to generate a new, random token.
        Setting to an empty string disables authentication altogether, which is NOT RECOMMENDED.
        """)
    ).tag(config=True)
    _token_generated = True
    @default('token')
    def _token_default(self):
        if os.getenv('JUPYTER_TOKEN'):
            self._token_generated = False
            return os.getenv('JUPYTER_TOKEN')
        if os.getenv('JUPYTER_TOKEN_FILE'):
            self._token_generated = False
            with open(os.getenv('JUPYTER_TOKEN_FILE')) as token_file:
                return token_file.read()
        if self.password:
            # no token if password is enabled
View remainder of file in raw view
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

notebookapp.py

notebookapp.py

Files

notebookapp.py

Latest commit

History

notebookapp.py

File metadata and controls
