mirror of
https://github.com/netbox-community/netbox.git
synced 2025-07-16 20:22:53 -06:00
244 lines
8.6 KiB
Python
244 lines
8.6 KiB
Python
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 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: 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 # 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
|