netbox/netbox/core/plugins.py

import datetime
import importlib
import importlib.util
from dataclasses import dataclass, field
from typing import Optional

import requests
from django.conf import settings
from django.core.cache import cache
from django.utils.translation import gettext_lazy as _

from netbox.plugins import PluginConfig
from utilities.datetime import datetime_from_timestamp

USER_AGENT_STRING = f'NetBox/{settings.RELEASE.version} {settings.RELEASE.edition}'


@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 = ''
    status: str = ''
    title_short: str = ''
    title_long: str = ''
    tag_line: str = ''
    description_short: str = ''
    slug: str = ''
    author: Optional[PluginAuthor] = 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  # extra field for locally installed plugins
    is_installed: bool = False
    installed_version: str = ''


def get_local_plugins():
    """
    Return a dictionary of all locally-installed plugins, mapped by name.
    """
    plugins = {}
    for plugin_name in settings.PLUGINS:
        plugin = importlib.import_module(plugin_name)
        plugin_config: PluginConfig = plugin.config

        plugins[plugin_config.name] = Plugin(
            slug=plugin_config.name,
            title_short=plugin_config.verbose_name,
            tag_line=plugin_config.description,
            description_short=plugin_config.description,
            is_local=True,
            is_installed=True,
            installed_version=plugin_config.version,
        )

    return plugins


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

    def get_pages():
        # TODO: pagination is currently broken in API
        payload = {'page': '1', 'per_page': '50'}
        first_page = session.get(
            settings.PLUGIN_CATALOG_URL,
            headers={'User-Agent': USER_AGENT_STRING},
            proxies=settings.HTTP_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=settings.HTTP_PROXIES,
                timeout=3,
                params=payload
            ).json()
            yield next_page

    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['slug']] = Plugin(
                id=data['id'],
                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


def get_plugins():
    """
    Return a dictionary of all plugins (both catalog and locally installed), mapped by name.
    """
    local_plugins = get_local_plugins()
    catalog_plugins = cache.get('plugins-catalog-feed')
    if not catalog_plugins:
        catalog_plugins = get_catalog_plugins()
        cache.set('plugins-catalog-feed', catalog_plugins, 3600)

    plugins = catalog_plugins
    for k, v in local_plugins.items():
        if k in plugins:
            plugins[k].is_local = True
            plugins[k].is_installed = True
        else:
            plugins[k] = v

    return plugins