"""Base classes for the extension manager.""" # Copyright (c) Jupyter Development Team. # Distributed under the terms of the Modified BSD License. import json import re from dataclasses import dataclass, field, fields, replace from pathlib import Path from typing import Dict, FrozenSet, List, Optional, Set, Tuple, Union import tornado from jupyterlab_server.translation_utils import translator from traitlets import Enum from traitlets.config import Configurable, LoggingConfigurable from jupyterlab.commands import ( _AppHandler, _ensure_options, disable_extension, enable_extension, get_app_info, ) PYTHON_TO_SEMVER = {"a": "-alpha.", "b": "-beta.", "rc": "-rc."} def _ensure_compat_errors(info, app_options): """Ensure that the app info has compat_errors field""" handler = _AppHandler(app_options) info["compat_errors"] = handler._get_extension_compat() _message_map = { "install": re.compile(r"(?P.*) needs to be included in build"), "uninstall": re.compile(r"(?P.*) needs to be removed from build"), "update": re.compile(r"(?P.*) changed from (?P.*) to (?P.*)"), } def _build_check_info(app_options): """Get info about packages scheduled for (un)install/update""" handler = _AppHandler(app_options) messages = handler.build_check(fast=True) # Decode the messages into a dict: status = {"install": [], "uninstall": [], "update": []} for msg in messages: for key, pattern in _message_map.items(): match = pattern.match(msg) if match: status[key].append(match.group("name")) return status @dataclass(frozen=True) class ExtensionPackage: """Extension package entry. Attributes: name: Package name description: Package description homepage_url: Package home page pkg_type: Type of package - ["prebuilt", "source"] allowed: [optional] Whether this extension is allowed or not - default True approved: [optional] Whether the package is approved by your administrators - default False companion: [optional] Type of companion for the frontend extension - [None, "kernel", "server"]; default None core: [optional] Whether the package is a core package or not - default False enabled: [optional] Whether the package is enabled or not - default False install: [optional] Extension package installation instructions - default None installed: [optional] Whether the extension is currently installed - default None installed_version: [optional] Installed version - default "" latest_version: [optional] Latest available version - default "" status: [optional] Package status - ["ok", "warning", "error"]; default "ok" author: [optional] Package author - default None license: [optional] Package license - default None bug_tracker_url: [optional] Package bug tracker URL - default None documentation_url: [optional] Package documentation URL - default None package_manager_url: Package home page in the package manager - default None repository_url: [optional] Package code repository URL - default None """ name: str description: str homepage_url: str pkg_type: str allowed: bool = True approved: bool = False companion: Optional[str] = None core: bool = False enabled: bool = False install: Optional[dict] = None installed: Optional[bool] = None installed_version: str = "" latest_version: str = "" status: str = "ok" author: Optional[str] = None license: Optional[str] = None bug_tracker_url: Optional[str] = None documentation_url: Optional[str] = None package_manager_url: Optional[str] = None repository_url: Optional[str] = None @dataclass(frozen=True) class ActionResult: """Action result Attributes: status: Action status - ["ok", "warning", "error"] message: Action status explanation needs_restart: Required action follow-up - Valid follow-up are "frontend", "kernel" and "server" """ # Note: no simple way to use Enum in dataclass - https://stackoverflow.com/questions/72859557/typing-dataclass-that-can-only-take-enum-values # keeping str for simplicity status: str message: Optional[str] = None needs_restart: List[str] = field(default_factory=list) @dataclass(frozen=True) class PluginManagerOptions: """Plugin manager options. Attributes: lock_all: Whether to lock (prevent enabling/disabling) all plugins. lock_rules: A list of plugins or extensions that cannot be toggled. If extension name is provided, all its plugins will be disabled. The plugin names need to follow colon-separated format of `extension:plugin`. """ lock_rules: FrozenSet[str] = field(default_factory=frozenset) lock_all: bool = False @dataclass(frozen=True) class ExtensionManagerOptions(PluginManagerOptions): """Extension manager options. Attributes: allowed_extensions_uris: A list of comma-separated URIs to get the allowed extensions list blocked_extensions_uris: A list of comma-separated URIs to get the blocked extensions list listings_refresh_seconds: The interval delay in seconds to refresh the lists listings_tornado_options: The optional kwargs to use for the listings HTTP requests as described on https://www.tornadoweb.org/en/stable/httpclient.html#tornado.httpclient.HTTPRequest """ allowed_extensions_uris: Set[str] = field(default_factory=set) blocked_extensions_uris: Set[str] = field(default_factory=set) listings_refresh_seconds: int = 60 * 60 listings_tornado_options: dict = field(default_factory=dict) @dataclass(frozen=True) class ExtensionManagerMetadata: """Extension manager metadata. Attributes: name: Extension manager name to be displayed can_install: Whether the extension manager can un-/install packages (default False) install_path: Installation path for the extensions (default None); e.g. environment path """ name: str can_install: bool = False install_path: Optional[str] = None @dataclass class ExtensionsCache: """Extensions cache Attributes: cache: Extension list per page last_page: Last available page result """ cache: Dict[int, Optional[Dict[str, ExtensionPackage]]] = field(default_factory=dict) last_page: int = 1 class PluginManager(LoggingConfigurable): """Plugin manager enables or disables plugins unless locked. It can also disable/enable all plugins in an extension. Args: app_options: Application options ext_options: Plugin manager (subset of extension manager) options parent: Configurable parent Attributes: app_options: Application options options: Plugin manager options """ level = Enum( values=["sys_prefix", "user", "system"], default_value="sys_prefix", help="Level at which to manage plugins: sys_prefix, user, system", ).tag(config=True) def __init__( self, app_options: Optional[dict] = None, ext_options: Optional[dict] = None, parent: Optional[Configurable] = None, ) -> None: super().__init__(parent=parent) self.log.debug( "Plugins in %s will managed on the %s level", self.__class__.__name__, self.level ) self.app_options = _ensure_options(app_options) plugin_options_field = {f.name for f in fields(PluginManagerOptions)} plugin_options = { option: value for option, value in (ext_options or {}).items() if option in plugin_options_field } self.options = PluginManagerOptions(**plugin_options) async def plugin_locks(self) -> dict: """Get information about locks on plugin enabling/disabling""" return { "lockRules": list(self.options.lock_rules), "allLocked": self.options.lock_all, } def _find_locked(self, plugins_or_extensions: List[str]) -> FrozenSet[str]: """Find a subset of plugins (or extensions) which are locked""" if self.options.lock_all: return set(plugins_or_extensions) locked_subset = set() extensions_with_locked_plugins = { plugin.split(":")[0] for plugin in self.options.lock_rules } for plugin in plugins_or_extensions: if ":" in plugin: # check directly if this is a plugin identifier (has colon) if plugin in self.options.lock_rules: locked_subset.add(plugin) elif plugin in extensions_with_locked_plugins: # this is an extension - we need to check for >any< plugin # belonging to said extension locked_subset.add(plugin) return locked_subset async def disable(self, plugins: Union[str, List[str]]) -> ActionResult: """Disable a set of plugins (or an extension). Args: plugins: The list of plugins to disable Returns: The action result """ plugins = plugins if isinstance(plugins, list) else [plugins] locked = self._find_locked(plugins) trans = translator.load("jupyterlab") if locked: return ActionResult( status="error", message=trans.gettext( "The following plugins cannot be disabled as they are locked: " ) + ", ".join(locked), ) try: for plugin in plugins: disable_extension(plugin, app_options=self.app_options, level=self.level) return ActionResult(status="ok", needs_restart=["frontend"]) except Exception as err: return ActionResult(status="error", message=repr(err)) async def enable(self, plugins: Union[str, List[str]]) -> ActionResult: """Enable a set of plugins (or an extension). Args: plugins: The list of plugins to enable Returns: The action result """ plugins = plugins if isinstance(plugins, list) else [plugins] locked = self._find_locked(plugins) trans = translator.load("jupyterlab") if locked: return ActionResult( status="error", message=trans.gettext( "The following plugins cannot be enabled as they are locked: " ) + ", ".join(locked), ) try: for plugin in plugins: enable_extension(plugin, app_options=self.app_options, level=self.level) return ActionResult(status="ok", needs_restart=["frontend"]) except Exception as err: return ActionResult(status="error", message=repr(err)) class ExtensionManager(PluginManager): """Base abstract extension manager. Note: Any concrete implementation will need to implement the five following abstract methods: - :ref:`metadata` - :ref:`get_latest_version` - :ref:`list_packages` - :ref:`install` - :ref:`uninstall` It could be interesting to override the :ref:`get_normalized_name` method too. Args: app_options: Application options ext_options: Extension manager options parent: Configurable parent Attributes: log: Logger app_dir: Application directory core_config: Core configuration app_options: Application options options: Extension manager options """ def __init__( self, app_options: Optional[dict] = None, ext_options: Optional[dict] = None, parent: Optional[Configurable] = None, ) -> None: super().__init__(app_options=app_options, ext_options=ext_options, parent=parent) self.log = self.app_options.logger self.app_dir = Path(self.app_options.app_dir) self.core_config = self.app_options.core_config self.options = ExtensionManagerOptions(**(ext_options or {})) self._extensions_cache: Dict[Optional[str], ExtensionsCache] = {} self._listings_cache: Optional[dict] = None self._listings_block_mode = True self._listing_fetch: Optional[tornado.ioloop.PeriodicCallback] = None if len(self.options.allowed_extensions_uris) or len(self.options.blocked_extensions_uris): self._listings_block_mode = len(self.options.allowed_extensions_uris) == 0 if not self._listings_block_mode and len(self.options.blocked_extensions_uris) > 0: self.log.warning( "You have define simultaneously blocked and allowed extensions listings. The allowed listing will take precedence." ) self._listing_fetch = tornado.ioloop.PeriodicCallback( self._fetch_listings, callback_time=self.options.listings_refresh_seconds * 1000, jitter=0.1, ) self._listing_fetch.start() def __del__(self): if self._listing_fetch is not None: self._listing_fetch.stop() @property def metadata(self) -> ExtensionManagerMetadata: """Extension manager metadata.""" raise NotImplementedError() async def get_latest_version(self, extension: str) -> Optional[str]: """Return the latest available version for a given extension. Args: pkg: The extension name Returns: The latest available version """ raise NotImplementedError() async def list_packages( self, query: str, page: int, per_page: int ) -> Tuple[Dict[str, ExtensionPackage], Optional[int]]: """List the available extensions. Args: query: The search extension query page: The result page per_page: The number of results per page Returns: The available extensions in a mapping {name: metadata} The results last page; None if the manager does not support pagination """ raise NotImplementedError() async def install(self, extension: str, version: Optional[str] = None) -> ActionResult: """Install the required extension. Note: If the user must be notified with a message (like asking to restart the server), the result should be {"status": "warning", "message": ""} Args: extension: The extension name version: The version to install; default None (i.e. the latest possible) Returns: The action result """ raise NotImplementedError() async def uninstall(self, extension: str) -> ActionResult: """Uninstall the required extension. Note: If the user must be notified with a message (like asking to restart the server), the result should be {"status": "warning", "message": ""} Args: extension: The extension name Returns: The action result """ raise NotImplementedError() @staticmethod def get_semver_version(version: str) -> str: """Convert a Python version to Semver version. It: - drops ``.devN`` and ``.postN`` - converts ``aN``, ``bN`` and ``rcN`` to ``-alpha.N``, ``-beta.N``, ``-rc.N`` respectively Args: version: Version to convert Returns Semver compatible version """ return re.sub( r"(a|b|rc)(\d+)$", lambda m: f"{PYTHON_TO_SEMVER[m.group(1)]}{m.group(2)}", re.subn(r"\.(dev|post)\d+", "", version)[0], ) def get_normalized_name(self, extension: ExtensionPackage) -> str: """Normalize extension name. Extension have multiple parts, npm package, Python package,... Sub-classes may override this method to ensure the name of an extension from the service provider and the local installed listing is matching. Args: extension: The extension metadata Returns: The normalized name """ return extension.name async def list_extensions( self, query: Optional[str] = None, page: int = 1, per_page: int = 30 ) -> Tuple[List[ExtensionPackage], Optional[int]]: """List extensions for a given ``query`` search term. This will return the extensions installed (if ``query`` is None) or available if allowed by the listing settings. Args: query: [optional] Query search term. Returns: The extensions Last page of results """ if query not in self._extensions_cache or page not in self._extensions_cache[query].cache: await self.refresh(query, page, per_page) # filter using listings settings if self._listings_cache is None and self._listing_fetch is not None: await self._listing_fetch.callback() cache = self._extensions_cache[query].cache[page] if cache is None: cache = {} extensions = list(cache.values()) if query is not None and self._listings_cache is not None: listing = list(self._listings_cache) extensions = [] if self._listings_block_mode: for name, ext in cache.items(): if name not in listing: extensions.append(replace(ext, allowed=True)) elif ext.installed_version: self.log.warning(f"Blocked extension '{name}' is installed.") extensions.append(replace(ext, allowed=False)) else: for name, ext in cache.items(): if name in listing: extensions.append(replace(ext, allowed=True)) elif ext.installed_version: self.log.warning(f"Not allowed extension '{name}' is installed.") extensions.append(replace(ext, allowed=False)) return extensions, self._extensions_cache[query].last_page async def refresh(self, query: Optional[str], page: int, per_page: int) -> None: """Refresh the list of extensions.""" if query in self._extensions_cache: self._extensions_cache[query].cache[page] = None await self._update_extensions_list(query, page, per_page) async def _fetch_listings(self) -> None: """Fetch the listings for the extension manager.""" rules = [] client = tornado.httpclient.AsyncHTTPClient() if self._listings_block_mode: if len(self.options.blocked_extensions_uris): self.log.info( f"Fetching blocked extensions from {self.options.blocked_extensions_uris}" ) for blocked_extensions_uri in self.options.blocked_extensions_uris: r = await client.fetch( blocked_extensions_uri, **self.options.listings_tornado_options, ) j = json.loads(r.body) rules.extend(j.get("blocked_extensions", [])) elif len(self.options.allowed_extensions_uris): self.log.info( f"Fetching allowed extensions from { self.options.allowed_extensions_uris}" ) for allowed_extensions_uri in self.options.allowed_extensions_uris: r = await client.fetch( allowed_extensions_uri, **self.options.listings_tornado_options, ) j = json.loads(r.body) rules.extend(j.get("allowed_extensions", [])) self._listings_cache = {r["name"]: r for r in rules} async def _get_installed_extensions( self, get_latest_version=True ) -> Dict[str, ExtensionPackage]: """Get the installed extensions. Args: get_latest_version: Whether to fetch the latest extension version or not. Returns: The installed extensions as a mapping {name: metadata} """ app_options = self.app_options info = get_app_info(app_options=app_options) build_check_info = _build_check_info(app_options) _ensure_compat_errors(info, app_options) extensions = {} # TODO: the three for-loops below can be run concurrently for name, data in info["federated_extensions"].items(): status = "ok" pkg_info = data if info["compat_errors"].get(name, None): status = "error" normalized_name = self._normalize_name(name) pkg = ExtensionPackage( name=normalized_name, description=pkg_info.get("description", ""), homepage_url=data.get("url", ""), enabled=(name not in info["disabled"]), core=False, latest_version=ExtensionManager.get_semver_version(data["version"]), installed=True, installed_version=ExtensionManager.get_semver_version(data["version"]), status=status, install=data.get("install", {}), pkg_type="prebuilt", companion=self._get_companion(data), author=data.get("author", {}).get("name", data.get("author")), license=data.get("license"), bug_tracker_url=data.get("bugs", {}).get("url"), repository_url=data.get("repository", {}).get("url", data.get("repository")), ) if get_latest_version: pkg = replace(pkg, latest_version=await self.get_latest_version(pkg.name)) extensions[normalized_name] = pkg for name, data in info["extensions"].items(): if name in info["shadowed_exts"]: continue status = "ok" if info["compat_errors"].get(name, None): status = "error" else: for packages in build_check_info.values(): if name in packages: status = "warning" normalized_name = self._normalize_name(name) pkg = ExtensionPackage( name=normalized_name, description=data.get("description", ""), homepage_url=data["url"], enabled=(name not in info["disabled"]), core=False, latest_version=ExtensionManager.get_semver_version(data["version"]), installed=True, installed_version=ExtensionManager.get_semver_version(data["version"]), status=status, pkg_type="source", companion=self._get_companion(data), author=data.get("author", {}).get("name", data.get("author")), license=data.get("license"), bug_tracker_url=data.get("bugs", {}).get("url"), repository_url=data.get("repository", {}).get("url", data.get("repository")), ) if get_latest_version: pkg = replace(pkg, latest_version=await self.get_latest_version(pkg.name)) extensions[normalized_name] = pkg for name in build_check_info["uninstall"]: data = self._get_scheduled_uninstall_info(name) if data is not None: normalized_name = self._normalize_name(name) pkg = ExtensionPackage( name=normalized_name, description=data.get("description", ""), homepage_url=data.get("homepage", ""), installed=False, enabled=False, core=False, latest_version=ExtensionManager.get_semver_version(data["version"]), installed_version=ExtensionManager.get_semver_version(data["version"]), status="warning", pkg_type="prebuilt", author=data.get("author", {}).get("name", data.get("author")), license=data.get("license"), bug_tracker_url=data.get("bugs", {}).get("url"), repository_url=data.get("repository", {}).get("url", data.get("repository")), ) extensions[normalized_name] = pkg return extensions def _get_companion(self, data: dict) -> Optional[str]: companion = None if "discovery" in data["jupyterlab"]: if "server" in data["jupyterlab"]["discovery"]: companion = "server" elif "kernel" in data["jupyterlab"]["discovery"]: companion = "kernel" return companion def _get_scheduled_uninstall_info(self, name) -> Optional[dict]: """Get information about a package that is scheduled for uninstallation""" target = self.app_dir / "staging" / "node_modules" / name / "package.json" if target.exists(): with target.open() as fid: return json.load(fid) else: return None def _normalize_name(self, name: str) -> str: """Normalize extension name; by default does nothing. Args: name: Extension name Returns: Normalized name """ return name async def _update_extensions_list( self, query: Optional[str] = None, page: int = 1, per_page: int = 30 ) -> None: """Update the list of extensions""" last_page = None if query is not None: # Get the available extensions extensions, last_page = await self.list_packages(query, page, per_page) else: # Get the installed extensions extensions = await self._get_installed_extensions() if query in self._extensions_cache: self._extensions_cache[query].cache[page] = extensions self._extensions_cache[query].last_page = last_page or 1 else: self._extensions_cache[query] = ExtensionsCache({page: extensions}, last_page or 1)