netbox/netbox/core/plugins.py

import datetime
import importlib
from dataclasses import dataclass, field

import requests
from django.conf import settings
from django.core.cache import cache

from netbox.plugins import PluginConfig
from netbox.registry import registry
from utilities.datetime import datetime_from_timestamp
from utilities.proxy import resolve_proxies

USER_AGENT_STRING = f'NetBox/{settings.RELEASE.version} {settings.RELEASE.edition}'
CACHE_KEY_CATALOG_FEED = 'plugins-catalog-feed'


@dataclass
class PluginAuthor:
    """
    Identifying information for the author of a plugin.
    """
    name: str
    org_id: str = ''
    url: str = ''


@dataclass
class PluginVersion:
    """
    Details for a specific versioned release of a plugin.
    """
    date: datetime.datetime = None
    version: str = ''
    netbox_min_version: str = ''
    netbox_max_version: str = ''
    has_model: bool = False
    is_certified: bool = False
    is_feature: bool = False
    is_integration: bool = False
    is_netboxlabs_supported: bool = False


@dataclass
class Plugin:
    """
    The representation of a NetBox plugin in the catalog API.
    """
    id: str = ''
    icon_url: str = ''
    status: str = ''
    title_short: str = ''
    title_long: str = ''
    tag_line: str = ''
    description_short: str = ''
    slug: str = ''
    author: PluginAuthor | None = None
    created_at: datetime.datetime = None
    updated_at: datetime.datetime = None
    license_type: str = ''
    homepage_url: str = ''
    package_name_pypi: str = ''
    config_name: str = ''
    is_certified: bool = False
    release_latest: PluginVersion = field(default_factory=PluginVersion)
    release_recent_history: list[PluginVersion] = field(default_factory=list)
    is_local: bool = False  # Indicates that the plugin is listed in settings.PLUGINS (i.e. installed)
    is_loaded: bool = False  # Indicates whether the plugin successfully loaded at launch
    installed_version: str = ''
    netbox_min_version: str = ''
    netbox_max_version: str = ''


def get_local_plugins(plugins=None):
    """
    Return a dictionary of all locally-installed plugins, mapped by name.
    """
    plugins = plugins or {}
    local_plugins = {}

    # Gather all locally-installed plugins
    for plugin_name in settings.PLUGINS:
        plugin = importlib.import_module(plugin_name)
        plugin_config: PluginConfig = plugin.config
        installed_version = plugin_config.version
        if plugin_config.release_track:
            installed_version = f'{installed_version}-{plugin_config.release_track}'

        if plugin_config.author:
            author = PluginAuthor(
                name=plugin_config.author,
            )
        else:
            author = None

        local_plugins[plugin_config.name] = Plugin(
            config_name=plugin_config.name,
            title_short=plugin_config.verbose_name,
            title_long=plugin_config.verbose_name,
            tag_line=plugin_config.description,
            description_short=plugin_config.description,
            is_local=True,
            is_loaded=plugin_name in registry['plugins']['installed'],
            installed_version=installed_version,
            netbox_min_version=plugin_config.min_version,
            netbox_max_version=plugin_config.max_version,
            author=author,
        )

    # Update catalog entries for local plugins, or add them to the list if not listed
    for k, v in local_plugins.items():
        if k in plugins:
            plugins[k].is_local = v.is_local
            plugins[k].is_loaded = v.is_loaded
            plugins[k].installed_version = v.installed_version
        else:
            plugins[k] = v

    # Update plugin table config for hidden and static plugins
    hidden = settings.PLUGINS_CATALOG_CONFIG.get('hidden', [])
    static = settings.PLUGINS_CATALOG_CONFIG.get('static', [])
    for k, v in plugins.items():
        v.hidden = k in hidden
        v.static = k in static

    return plugins


def get_catalog_plugins():
    """
    Return a dictionary of all entries in the plugins catalog, mapped by name.
    """
    session = requests.Session()

    # Disable catalog fetching for isolated deployments
    if settings.ISOLATED_DEPLOYMENT:
        return {}

    def get_pages():
        # TODO: pagination is currently broken in API
        payload = {'page': '1', 'per_page': '50'}
        proxies = resolve_proxies(url=settings.PLUGIN_CATALOG_URL)
        first_page = session.get(
            settings.PLUGIN_CATALOG_URL,
            headers={'User-Agent': USER_AGENT_STRING},
            proxies=proxies,
            timeout=3,
            params=payload
        ).json()
        yield first_page
        num_pages = first_page['metadata']['pagination']['last_page']

        for page in range(2, num_pages + 1):
            payload['page'] = page
            next_page = session.get(
                settings.PLUGIN_CATALOG_URL,
                headers={'User-Agent': USER_AGENT_STRING},
                proxies=proxies,
                timeout=3,
                params=payload
            ).json()
            yield next_page

    def make_plugin_dict():
        plugins = {}

        for page in get_pages():
            for data in page['data']:

                # Populate releases
                releases = []
                for version in data['release_recent_history']:
                    releases.append(
                        PluginVersion(
                            date=datetime_from_timestamp(version['date']),
                            version=version['version'],
                            netbox_min_version=version['netbox_min_version'],
                            netbox_max_version=version['netbox_max_version'],
                            has_model=version['has_model'],
                            is_certified=version['is_certified'],
                            is_feature=version['is_feature'],
                            is_integration=version['is_integration'],
                            is_netboxlabs_supported=version['is_netboxlabs_supported'],
                        )
                    )
                releases = sorted(releases, key=lambda x: x.date, reverse=True)
                latest_release = PluginVersion(
                    date=datetime_from_timestamp(data['release_latest']['date']),
                    version=data['release_latest']['version'],
                    netbox_min_version=data['release_latest']['netbox_min_version'],
                    netbox_max_version=data['release_latest']['netbox_max_version'],
                    has_model=data['release_latest']['has_model'],
                    is_certified=data['release_latest']['is_certified'],
                    is_feature=data['release_latest']['is_feature'],
                    is_integration=data['release_latest']['is_integration'],
                    is_netboxlabs_supported=data['release_latest']['is_netboxlabs_supported'],
                )

                # Populate author (if any)
                if data['author']:
                    author = PluginAuthor(
                        name=data['author']['name'],
                        org_id=data['author']['org_id'],
                        url=data['author']['url'],
                    )
                else:
                    author = None

                # Populate plugin data
                plugins[data['config_name']] = Plugin(
                    id=data['id'],
                    icon_url=data['icon'],
                    status=data['status'],
                    title_short=data['title_short'],
                    title_long=data['title_long'],
                    tag_line=data['tag_line'],
                    description_short=data['description_short'],
                    slug=data['slug'],
                    author=author,
                    created_at=datetime_from_timestamp(data['created_at']),
                    updated_at=datetime_from_timestamp(data['updated_at']),
                    license_type=data['license_type'],
                    homepage_url=data['homepage_url'],
                    package_name_pypi=data['package_name_pypi'],
                    config_name=data['config_name'],
                    is_certified=data['is_certified'],
                    release_latest=latest_release,
                    release_recent_history=releases,
                )

        return plugins

    catalog_plugins = cache.get(CACHE_KEY_CATALOG_FEED, default={})
    if not catalog_plugins:
        try:
            catalog_plugins = make_plugin_dict()
            cache.set(CACHE_KEY_CATALOG_FEED, catalog_plugins, 3600)
        except requests.exceptions.RequestException:
            pass

    return catalog_plugins