Coverage for core/plugins.py: 39%

1import datetime 

2import importlib 

3from dataclasses import dataclass, field 

4 

5import requests 

6from django.conf import settings 

7from django.core.cache import cache 

8 

9from netbox.plugins import PluginConfig 

10from netbox.registry import registry 

11from utilities.datetime import datetime_from_timestamp 

12from utilities.proxy import resolve_proxies 

13 

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

15CACHE_KEY_CATALOG_FEED = 'plugins-catalog-feed' 

16 

17 

18@dataclass 

19class PluginAuthor: 

20 """ 

21 Identifying information for the author of a plugin. 

22 """ 

23 name: str 

24 org_id: str = '' 

25 url: str = '' 

26 

27 

28@dataclass 

29class PluginVersion: 

30 """ 

31 Details for a specific versioned release of a plugin. 

32 """ 

33 date: datetime.datetime = None 

34 version: str = '' 

35 netbox_min_version: str = '' 

36 netbox_max_version: str = '' 

37 has_model: bool = False 

38 is_certified: bool = False 

39 is_feature: bool = False 

40 is_integration: bool = False 

41 is_netboxlabs_supported: bool = False 

42 

43 

44@dataclass 

45class Plugin: 

46 """ 

47 The representation of a NetBox plugin in the catalog API. 

48 """ 

49 id: str = '' 

50 icon_url: str = '' 

51 status: str = '' 

52 title_short: str = '' 

53 title_long: str = '' 

54 tag_line: str = '' 

55 description_short: str = '' 

56 slug: str = '' 

57 author: PluginAuthor | None = None 

58 created_at: datetime.datetime = None 

59 updated_at: datetime.datetime = None 

60 license_type: str = '' 

61 homepage_url: str = '' 

62 package_name_pypi: str = '' 

63 config_name: str = '' 

64 is_certified: bool = False 

65 release_latest: PluginVersion = field(default_factory=PluginVersion) 

66 release_recent_history: list[PluginVersion] = field(default_factory=list) 

67 is_local: bool = False # Indicates that the plugin is listed in settings.PLUGINS (i.e. installed) 

68 is_loaded: bool = False # Indicates whether the plugin successfully loaded at launch 

69 installed_version: str = '' 

70 netbox_min_version: str = '' 

71 netbox_max_version: str = '' 

72 

73 

74def get_local_plugins(plugins=None): 

75 """ 

76 Return a dictionary of all locally-installed plugins, mapped by name. 

77 """ 

78 plugins = plugins or {} 

79 local_plugins = {} 

80 

81 # Gather all locally-installed plugins 

82 for plugin_name in settings.PLUGINS: 

83 plugin = importlib.import_module(plugin_name) 

84 plugin_config: PluginConfig = plugin.config 

85 installed_version = plugin_config.version 

86 if plugin_config.release_track: 

87 installed_version = f'{installed_version}-{plugin_config.release_track}' 

88 

89 if plugin_config.author: 

90 author = PluginAuthor( 

91 name=plugin_config.author, 

92 ) 

93 else: 

94 author = None 

95 

96 local_plugins[plugin_config.name] = Plugin( 

97 config_name=plugin_config.name, 

98 title_short=plugin_config.verbose_name, 

99 title_long=plugin_config.verbose_name, 

100 tag_line=plugin_config.description, 

101 description_short=plugin_config.description, 

102 is_local=True, 

103 is_loaded=plugin_name in registry['plugins']['installed'], 

104 installed_version=installed_version, 

105 netbox_min_version=plugin_config.min_version, 

106 netbox_max_version=plugin_config.max_version, 

107 author=author, 

108 ) 

109 

110 # Update catalog entries for local plugins, or add them to the list if not listed 

111 for k, v in local_plugins.items(): 

112 if k in plugins: 

113 plugins[k].is_local = v.is_local 

114 plugins[k].is_loaded = v.is_loaded 

115 plugins[k].installed_version = v.installed_version 

116 else: 

117 plugins[k] = v 

118 

119 # Update plugin table config for hidden and static plugins 

120 hidden = settings.PLUGINS_CATALOG_CONFIG.get('hidden', []) 

121 static = settings.PLUGINS_CATALOG_CONFIG.get('static', []) 

122 for k, v in plugins.items(): 

123 v.hidden = k in hidden 

124 v.static = k in static 

125 

126 return plugins 

127 

128 

129def get_catalog_plugins(): 

130 """ 

131 Return a dictionary of all entries in the plugins catalog, mapped by name. 

132 """ 

133 session = requests.Session() 

134 

135 # Disable catalog fetching for isolated deployments 

136 if settings.ISOLATED_DEPLOYMENT: 

137 return {} 

138 

139 def get_pages(): 

140 # TODO: pagination is currently broken in API 

141 payload = {'page': '1', 'per_page': '50'} 

142 proxies = resolve_proxies(url=settings.PLUGIN_CATALOG_URL) 

143 first_page = session.get( 

144 settings.PLUGIN_CATALOG_URL, 

145 headers={'User-Agent': USER_AGENT_STRING}, 

146 proxies=proxies, 

147 timeout=3, 

148 params=payload 

149 ).json() 

150 yield first_page 

151 num_pages = first_page['metadata']['pagination']['last_page'] 

152 

153 for page in range(2, num_pages + 1): 

154 payload['page'] = page 

155 next_page = session.get( 

156 settings.PLUGIN_CATALOG_URL, 

157 headers={'User-Agent': USER_AGENT_STRING}, 

158 proxies=proxies, 

159 timeout=3, 

160 params=payload 

161 ).json() 

162 yield next_page 

163 

164 def make_plugin_dict(): 

165 plugins = {} 

166 

167 for page in get_pages(): 

168 for data in page['data']: 

169 

170 # Populate releases 

171 releases = [] 

172 for version in data['release_recent_history']: 

173 releases.append( 

174 PluginVersion( 

175 date=datetime_from_timestamp(version['date']), 

176 version=version['version'], 

177 netbox_min_version=version['netbox_min_version'], 

178 netbox_max_version=version['netbox_max_version'], 

179 has_model=version['has_model'], 

180 is_certified=version['is_certified'], 

181 is_feature=version['is_feature'], 

182 is_integration=version['is_integration'], 

183 is_netboxlabs_supported=version['is_netboxlabs_supported'], 

184 ) 

185 ) 

186 releases = sorted(releases, key=lambda x: x.date, reverse=True) 

187 latest_release = PluginVersion( 

188 date=datetime_from_timestamp(data['release_latest']['date']), 

189 version=data['release_latest']['version'], 

190 netbox_min_version=data['release_latest']['netbox_min_version'], 

191 netbox_max_version=data['release_latest']['netbox_max_version'], 

192 has_model=data['release_latest']['has_model'], 

193 is_certified=data['release_latest']['is_certified'], 

194 is_feature=data['release_latest']['is_feature'], 

195 is_integration=data['release_latest']['is_integration'], 

196 is_netboxlabs_supported=data['release_latest']['is_netboxlabs_supported'], 

197 ) 

198 

199 # Populate author (if any) 

200 if data['author']: 

201 author = PluginAuthor( 

202 name=data['author']['name'], 

203 org_id=data['author']['org_id'], 

204 url=data['author']['url'], 

205 ) 

206 else: 

207 author = None 

208 

209 # Populate plugin data 

210 plugins[data['config_name']] = Plugin( 

211 id=data['id'], 

212 icon_url=data['icon'], 

213 status=data['status'], 

214 title_short=data['title_short'], 

215 title_long=data['title_long'], 

216 tag_line=data['tag_line'], 

217 description_short=data['description_short'], 

218 slug=data['slug'], 

219 author=author, 

220 created_at=datetime_from_timestamp(data['created_at']), 

221 updated_at=datetime_from_timestamp(data['updated_at']), 

222 license_type=data['license_type'], 

223 homepage_url=data['homepage_url'], 

224 package_name_pypi=data['package_name_pypi'], 

225 config_name=data['config_name'], 

226 is_certified=data['is_certified'], 

227 release_latest=latest_release, 

228 release_recent_history=releases, 

229 ) 

230 

231 return plugins 

232 

233 catalog_plugins = cache.get(CACHE_KEY_CATALOG_FEED, default={}) 

234 if not catalog_plugins: 

235 try: 

236 catalog_plugins = make_plugin_dict() 

237 cache.set(CACHE_KEY_CATALOG_FEED, catalog_plugins, 3600) 

238 except requests.exceptions.RequestException: 

239 pass 

240 

241 return catalog_plugins