Coverage for src/backend/InvenTree/plugin/installer.py: 18%

1"""Install a plugin into the python virtual environment.""" 

2 

3import re 

4import subprocess 

5import sys 

6 

7from django.conf import settings 

8from django.core.exceptions import ValidationError 

9from django.utils.translation import gettext_lazy as _ 

10 

11import structlog 

12 

13import plugin.models 

14import plugin.staticfiles 

15from InvenTree.exceptions import log_error 

16 

17logger = structlog.get_logger('inventree') 

18 

19 

20def pip_command(*args): 

21 """Build and run a pip command using using the current python executable. 

22 

23 Returns: The output of the pip command 

24 

25 Raises: 

26 subprocess.CalledProcessError: If the pip command fails 

27 """ 

28 python = sys.executable 

29 

30 command = [python, '-m', 'pip'] 

31 

32 command.extend(args) 

33 

34 command = [str(x) for x in command] 

35 

36 logger.info('Running pip command: %s', ' '.join(command)) 

37 

38 return subprocess.check_output( 

39 command, cwd=settings.BASE_DIR.parent, stderr=subprocess.STDOUT 

40 ) 

41 

42 

43def handle_pip_error(error, path: str) -> list: 

44 """Raise a ValidationError when the pip command fails. 

45 

46 - Log the error to the database 

47 - Format the output from a pip command into a list of error messages. 

48 - Raise an appropriate error 

49 """ 

50 log_error(path, scope='pip') 

51 

52 output = error.output.decode('utf-8') 

53 

54 logger.error('Pip command failed: %s', output) 

55 

56 errors = [] 

57 

58 for line in output.split('\n'): 

59 line = line.strip() 

60 

61 if line: 

62 errors.append(line) 

63 

64 if len(errors) > 1: 

65 raise ValidationError(errors) 

66 else: 

67 raise ValidationError(errors[0]) 

68 

69 

70def get_install_info(packagename: str) -> dict: 

71 """Determine the install information for a particular package. 

72 

73 - Uses 'pip show' to determine the install location of a package. 

74 """ 

75 logger.debug('get_install_info: %s', packagename) 

76 

77 # Remove version information 

78 for c in '<>=!@ ': 

79 packagename = packagename.split(c)[0] 

80 

81 info = {} 

82 

83 try: 

84 result = pip_command('show', packagename) 

85 

86 output = result.decode('utf-8').split('\n') 

87 

88 for line in output: 

89 parts = line.split(':') 

90 

91 if len(parts) >= 2: 

92 key = str(parts[0].strip().lower().replace('-', '_')) 

93 value = str(parts[1].strip()) 

94 

95 info[key] = value 

96 

97 except subprocess.CalledProcessError as error: 

98 log_error('get_install_info', scope='pip') 

99 

100 output = error.output.decode('utf-8') 

101 info['error'] = output 

102 logger.exception('Plugin lookup failed: %s', output) 

103 except Exception: 

104 log_error('get_install_info', scope='pip') 

105 

106 return info 

107 

108 

109def plugins_file_hash(): 

110 """Return the file hash for the plugins file.""" 

111 import hashlib 

112 

113 pf = settings.PLUGIN_FILE 

114 

115 if not pf or not pf.exists(): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 return None 

117 

118 try: 

119 with pf.open('rb') as f: 

120 # Note: Once we support 3.11 as a minimum, we can use hashlib.file_digest 

121 return hashlib.sha256(f.read()).hexdigest() 

122 except Exception: 

123 log_error('plugins_file_hash', scope='plugins') 

124 return None 

125 

126 

127def install_plugins_file(): 

128 """Install plugins from the plugins file.""" 

129 logger.info('Installing plugins from plugins file') 

130 

131 pf = settings.PLUGIN_FILE 

132 

133 if not pf or not pf.exists(): 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 logger.warning('Plugin file %s does not exist', pf) 

135 return 

136 

137 cmd = ['install', '--disable-pip-version-check', '-U', '-r', str(pf)] 

138 

139 try: 

140 pip_command(*cmd) 

141 except subprocess.CalledProcessError as error: 

142 output = error.output.decode('utf-8') 

143 logger.exception('Plugin file installation failed: %s', output) 

144 log_error('install_plugins_file', scope='pip') 

145 return False 

146 except Exception as exc: 

147 logger.exception('Plugin file installation failed: %s', exc) 

148 log_error('install_plugins_file', scope='pip') 

149 return False 

150 

151 # Collect plugin static files 

152 try: 

153 plugin.staticfiles.collect_plugins_static_files() 

154 except Exception: 

155 log_error('collect_plugins_static_files', scope='plugins') 

156 

157 # At this point, the plugins file has been installed 

158 return True 

159 

160 

161def update_plugins_file(install_name, full_package=None, version=None, remove=False): 

162 """Add a plugin to the plugins file.""" 

163 if remove: 

164 logger.info('Removing plugin from plugins file: %s', install_name) 

165 else: 

166 logger.info('Adding plugin to plugins file: %s', install_name) 

167 

168 # If a full package name is provided, use that instead 

169 if full_package and full_package != install_name: 

170 new_value = full_package 

171 else: 

172 new_value = f'{install_name}=={version}' if version else install_name 

173 

174 pf = settings.PLUGIN_FILE 

175 

176 if not pf or not pf.exists(): 

177 logger.warning('Plugin file %s does not exist', pf) 

178 return 

179 

180 def compare_line(line: str): 

181 """Check if a line in the file matches the installname.""" 

182 return re.match(rf'^{install_name}[\s=@]', line.strip()) 

183 

184 # First, read in existing plugin file 

185 try: 

186 with pf.open(mode='r') as f: 

187 lines = f.readlines() 

188 except Exception as exc: 

189 logger.exception('Failed to read plugins file: %s', exc) 

190 log_error('update_plugins_file', scope='plugins') 

191 return 

192 

193 # Reconstruct output file 

194 output = [] 

195 

196 found = False 

197 

198 # Check if plugin is already in file 

199 for line in lines: 

200 # Ignore processing for any commented lines 

201 if line.strip().startswith('#'): 

202 output.append(line) 

203 continue 

204 

205 if compare_line(line): 

206 found = True 

207 if not remove: 

208 # Replace line with new install name 

209 output.append(new_value) 

210 else: 

211 output.append(line) 

212 

213 # Append plugin to file 

214 if not found and not remove: 

215 output.append(new_value) 

216 

217 # Write file back to disk 

218 try: 

219 with pf.open(mode='w') as f: 

220 for line in output: 

221 f.write(line) 

222 

223 if not line.endswith('\n'): 

224 f.write('\n') 

225 except Exception as exc: 

226 logger.exception('Failed to add plugin to plugins file: %s', exc) 

227 log_error('update_plugins_file', scope='plugins') 

228 

229 

230def install_plugin(url=None, packagename=None, user=None, version=None): 

231 """Install a plugin into the python virtual environment. 

232 

233 Args: 

234 packagename: Optional package name to install 

235 url: Optional URL to install from 

236 user: Optional user performing the installation 

237 version: Optional version specifier 

238 """ 

239 if user and not user.is_superuser: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true

240 raise ValidationError(_('Only superuser accounts can administer plugins')) 

241 

242 if settings.PLUGINS_INSTALL_DISABLED: 242 ↛ 245line 242 didn't jump to line 245 because the condition on line 242 was always true

243 raise ValidationError(_('Plugin installation is disabled')) 

244 

245 logger.info('install_plugin: %s, %s', url, packagename) 

246 

247 # build up the command 

248 install_name = ['install', '-U', '--disable-pip-version-check'] 

249 

250 full_pkg = '' 

251 

252 if url: 

253 # use custom registration / VCS 

254 if True in [ 

255 identifier in url for identifier in ['git+https', 'hg+https', 'svn+svn'] 

256 ]: 

257 # using a VCS provider 

258 full_pkg = f'{packagename}@{url}' if packagename else url 

259 elif url: 

260 install_name.append('-i') 

261 full_pkg = url 

262 elif packagename: 

263 full_pkg = packagename 

264 

265 elif packagename: 

266 # use pypi 

267 full_pkg = packagename 

268 

269 if version: 

270 full_pkg = f'{full_pkg}=={version}' 

271 

272 if not full_pkg: 

273 raise ValidationError(_('No package name or URL provided for installation')) 

274 

275 # Sanitize the package name for installation 

276 if any(c in full_pkg for c in ';&|`$()'): 

277 raise ValidationError(_('Invalid characters in package name or URL')) 

278 

279 install_name.append(full_pkg) 

280 

281 ret = {} 

282 

283 # Execute installation via pip 

284 try: 

285 result = pip_command(*install_name) 

286 

287 ret['result'] = ret['success'] = _('Installed plugin successfully') 

288 ret['output'] = str(result, 'utf-8') 

289 

290 if packagename and (info := get_install_info(packagename)): 

291 if path := info.get('location'): 

292 ret['result'] = _(f'Installed plugin into {path}') 

293 ret['version'] = info.get('version') 

294 

295 except subprocess.CalledProcessError as error: 

296 handle_pip_error(error, 'plugin_install') 

297 except Exception: 

298 log_error('install_plugin', scope='plugins') 

299 

300 if version := ret.get('version'): 

301 # Save plugin to plugins file 

302 update_plugins_file(packagename, full_package=full_pkg, version=version) 

303 

304 # Reload the plugin registry, to discover the new plugin 

305 from plugin.registry import registry 

306 

307 registry.reload_plugins(full_reload=True, force_reload=True, collect=True) 

308 

309 # Update static files 

310 plugin.staticfiles.collect_plugins_static_files() 

311 

312 return ret 

313 

314 

315def validate_package_plugin(cfg: plugin.models.PluginConfig, user=None): 

316 """Validate a package plugin for update or removal.""" 

317 if not cfg.plugin: 

318 raise ValidationError(_('Plugin was not found in registry')) 

319 

320 if not cfg.is_package(): 

321 raise ValidationError(_('Plugin is not a packaged plugin')) 

322 

323 if not cfg.package_name: 

324 raise ValidationError(_('Plugin package name not found')) 

325 

326 if user and not user.is_staff: 

327 raise ValidationError(_('Only staff users can administer plugins')) 

328 

329 

330def uninstall_plugin(cfg: plugin.models.PluginConfig, user=None, delete_config=True): 

331 """Uninstall a plugin from the python virtual environment. 

332 

333 - The plugin must not be active 

334 - The plugin must be a "package" and have a valid package name 

335 

336 Args: 

337 cfg: PluginConfig object 

338 user: User performing the uninstall 

339 delete_config: If True, delete the plugin configuration from the database 

340 """ 

341 from plugin.registry import registry 

342 

343 if user and not user.is_superuser: 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true

344 raise ValidationError(_('Only superuser accounts can administer plugins')) 

345 

346 if settings.PLUGINS_INSTALL_DISABLED: 346 ↛ 349line 346 didn't jump to line 349 because the condition on line 346 was always true

347 raise ValidationError(_('Plugin uninstalling is disabled')) 

348 

349 if cfg.active: 

350 raise ValidationError( 

351 _('Plugin cannot be uninstalled as it is currently active') 

352 ) 

353 

354 if cfg.is_mandatory(): # pragma: no cover 

355 # This is only an additional check, as mandatory plugins cannot be deactivated 

356 raise ValidationError( 

357 'INVE-E10' + _('Plugin cannot be uninstalled as it is mandatory') 

358 ) 

359 

360 if cfg.is_sample(): 

361 raise ValidationError( 

362 'INVE-E10' + _('Plugin cannot be uninstalled as it is a sample plugin') 

363 ) 

364 

365 if cfg.is_builtin(): 

366 raise ValidationError( 

367 'INVE-E10' + _('Plugin cannot be uninstalled as it is a built-in plugin') 

368 ) 

369 

370 if not cfg.is_installed(): 

371 raise ValidationError(_('Plugin is not installed')) 

372 

373 validate_package_plugin(cfg, user) 

374 package_name = cfg.package_name 

375 

376 pkg_info = get_install_info(package_name) 

377 

378 if path := pkg_info.get('location'): 

379 # Uninstall the plugin using pip 

380 logger.info('Uninstalling plugin: %s from %s', package_name, path) 

381 try: 

382 pip_command('uninstall', '-y', package_name) 

383 except subprocess.CalledProcessError as error: 

384 handle_pip_error(error, 'plugin_uninstall') 

385 except Exception: 

386 log_error('uninstall_plugin', scope='plugins') 

387 else: 

388 # No matching install target found 

389 raise ValidationError(_('Plugin installation not found')) 

390 

391 # Update the plugins file 

392 update_plugins_file(package_name, remove=True) 

393 

394 if delete_config: 

395 logger.info('Deleting plugin configuration from database: %s', cfg.key) 

396 # Remove the plugin configuration from the database 

397 cfg.delete() 

398 

399 # Remove static files associated with this plugin 

400 plugin.staticfiles.clear_plugin_static_files(cfg.key) 

401 

402 # Reload the plugin registry 

403 registry.reload_plugins(full_reload=True, force_reload=True, collect=True) 

404 

405 return {'result': _('Uninstalled plugin successfully'), 'success': True}