Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / plugins / __init__.py: 98%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Provides the public API for the Bijux CLI's plugin management service. 

5 

6This module acts as the public facade for the plugin management service layer. 

7It exposes a curated set of high-level functions for core plugin operations, 

8ensuring a stable and convenient API for the rest of the application. 

9 

10This module exposes explicit imports for commonly used registry helpers. 

11""" 

12 

13from __future__ import annotations 

14 

15from contextlib import suppress 

16import importlib.metadata as md 

17import importlib.util 

18import logging 

19import os 

20from pathlib import Path 

21import shutil 

22import subprocess # nosec B404 - controlled internal execution 

23import sys 

24from typing import Any, cast 

25 

26from packaging.specifiers import SpecifierSet 

27 

28from bijux_cli.core.errors import BijuxError 

29from bijux_cli.core.version import version as cli_version 

30from bijux_cli.plugins.contracts import RegistryProtocol 

31from bijux_cli.services.contracts import ObservabilityProtocol, TelemetryProtocol 

32 

33logger = logging.getLogger(__name__) 

34 

35 

36def _di() -> Any | None: 

37 """Safely retrieves the current DIContainer instance. 

38 

39 Returns: 

40 DIContainer | None: The current dependency injection container, or None 

41 if it is not available or an error occurs. 

42 """ 

43 from bijux_cli.core.di import DIContainer 

44 

45 try: 

46 return DIContainer.current() 

47 except Exception: 

48 return None 

49 

50 

51def _obs() -> ObservabilityProtocol | None: 

52 """Safely resolves the `ObservabilityProtocol` service. 

53 

54 Returns: 

55 ObservabilityProtocol | None: The observability service, or None if it 

56 cannot be resolved. 

57 """ 

58 di = _di() 

59 if not di: 

60 return None 

61 try: 

62 return cast(ObservabilityProtocol, di.resolve(ObservabilityProtocol)) 

63 except KeyError: 

64 return None 

65 

66 

67def _tel() -> TelemetryProtocol | None: 

68 """Safely resolves the `TelemetryProtocol` service. 

69 

70 Returns: 

71 TelemetryProtocol | None: The telemetry service, or None if it 

72 cannot be resolved. 

73 """ 

74 di = _di() 

75 if not di: 

76 return None 

77 try: 

78 return cast(TelemetryProtocol, di.resolve(TelemetryProtocol)) 

79 except KeyError: 

80 return None 

81 

82 

83def get_plugins_dir() -> Path: 

84 """Returns the directory that stores installed plugins. 

85 

86 The path is determined by the `BIJUXCLI_PLUGINS_DIR` environment variable 

87 if set, otherwise it falls back to a default location. The directory is 

88 created if it does not exist. 

89 

90 Returns: 

91 Path: The resolved path to the plugins' directory. 

92 """ 

93 env_path = os.environ.get("BIJUXCLI_PLUGINS_DIR") 

94 if env_path: 

95 plugins_dir = Path(env_path) 

96 else: 

97 from bijux_cli.infra.paths import PLUGINS_DIR 

98 

99 plugins_dir = PLUGINS_DIR 

100 plugins_dir = plugins_dir.expanduser() 

101 

102 if plugins_dir.exists() and plugins_dir.is_symlink(): 

103 return plugins_dir 

104 if plugins_dir.is_dir(): 

105 return plugins_dir.resolve() 

106 if plugins_dir.exists(): 

107 return plugins_dir.resolve() 

108 plugins_dir.mkdir(parents=True, exist_ok=True) 

109 return plugins_dir.resolve() 

110 

111 

112def load_plugin_config(name: str) -> dict[str, Any]: 

113 """Loads a plugin's `config.yaml` file. 

114 

115 This function looks for a `config.yaml` file in the specified plugin's 

116 directory. It returns an empty dictionary if the file is missing. 

117 

118 Args: 

119 name (str): The name of the plugin whose config should be loaded. 

120 

121 Returns: 

122 dict[str, Any]: The plugin's configuration as a dictionary. 

123 

124 Raises: 

125 BijuxError: If the `PyYAML` library is not installed or if the file 

126 is corrupt and cannot be parsed. 

127 """ 

128 try: 

129 import yaml 

130 except ModuleNotFoundError as exc: 

131 raise BijuxError("PyYAML is required to read plugin configs") from exc 

132 

133 cfg_path = get_plugins_dir() / name / "config.yaml" 

134 if not cfg_path.exists(): 

135 return {} 

136 

137 try: 

138 data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) or {} 

139 except Exception as exc: 

140 tel = _tel() 

141 if tel: 

142 tel.event("plugin_config_failed", {"name": name, "error": str(exc)}) 

143 raise BijuxError(f"Failed to load config for '{name}': {exc}") from exc 

144 

145 obs = _obs() 

146 tel = _tel() 

147 if obs: 

148 obs.log("info", "Loaded plugin config", extra={"name": name}) 

149 if tel: 

150 tel.event("plugin_config_loaded", {"name": name}) 

151 return data 

152 

153 

154def verify_plugin_signature(path: Path, public_key: str | None) -> bool: 

155 """Verifies the signature of a plugin file. 

156 

157 Note: 

158 This is a placeholder function. Actual cryptographic verification is 

159 not implemented. 

160 

161 Args: 

162 path (Path): The path to the plugin file to verify. 

163 public_key (str | None): The public key to use for verification. 

164 

165 Returns: 

166 bool: True if the signature is "verified", False if no signature exists. 

167 

168 Raises: 

169 BijuxError: If a signature file exists but no public key is provided. 

170 """ 

171 sig = path.with_suffix(path.suffix + ".sig") 

172 tel = _tel() 

173 

174 if not sig.exists(): 

175 if tel: 

176 tel.event("plugin_unsigned", {"path": str(path)}) 

177 return False 

178 

179 if public_key is None: 

180 raise BijuxError(f"Signature found for {path} but no public_key provided") 

181 

182 if tel: 

183 tel.event("plugin_signature_verified", {"path": str(path)}) 

184 return True 

185 

186 

187def load_plugin( 

188 path: str | Path, 

189 module_name: str, 

190 *, 

191 public_key: str | None = None, 

192) -> Any: 

193 """Dynamically loads and instantiates a `Plugin` class from a `.py` file. 

194 

195 This function handles the dynamic import of a plugin's code, instantiates 

196 its main `Plugin` class, and performs compatibility checks. 

197 

198 Args: 

199 path (str | Path): The path to the `plugin.py` file. 

200 module_name (str): The name to assign to the imported module. 

201 public_key (str | None): An optional public key for signature verification. 

202 

203 Returns: 

204 Any: An instantiated object of the `Plugin` class from the file. 

205 

206 Raises: 

207 BijuxError: If the file is missing, the signature is invalid, the 

208 plugin is incompatible with the current CLI version, or an import 

209 error occurs. 

210 """ 

211 path = Path(path) 

212 if not path.is_file(): 

213 raise BijuxError(f"Plugin file not found: {path}") 

214 

215 if public_key: 

216 verify_plugin_signature(path, public_key) 

217 

218 spec = importlib.util.spec_from_file_location(module_name, str(path)) 

219 if not spec or not spec.loader: 

220 raise BijuxError(f"Cannot import plugin: {path}") 

221 

222 module = importlib.util.module_from_spec(spec) 

223 sys.modules[module_name] = module 

224 

225 try: 

226 spec.loader.exec_module(module) 

227 

228 plugin_class = getattr(module, "Plugin", None) 

229 if plugin_class is None: 

230 raise BijuxError("No `Plugin` class found in module") 

231 

232 plugin = plugin_class() 

233 

234 for target in (plugin_class, plugin): 

235 raw = getattr(target, "version", None) 

236 if raw is not None and not isinstance(raw, str): 

237 target.version = str(raw) 

238 

239 required = getattr(plugin, "requires_cli_version", f"=={cli_version}") 

240 spec_set = SpecifierSet(required) 

241 if not spec_set.contains(cli_version): 

242 raise BijuxError(f"Plugin requires CLI {required}, host is {cli_version}") 

243 

244 obs = _obs() 

245 cli_attr = getattr(plugin, "cli", None) 

246 if obs and (cli_attr is None or not callable(cli_attr)): 

247 obs.log( 

248 "warning", 

249 f"Plugin '{module_name}' has no callable `cli`", 

250 extra={"path": str(path)}, 

251 ) 

252 

253 tel = _tel() 

254 if tel: 

255 tel.event("plugin_loaded", {"name": getattr(plugin, "name", module_name)}) 

256 

257 return plugin 

258 

259 except Exception as exc: 

260 sys.modules.pop(module_name, None) 

261 raise BijuxError(f"Failed to load plugin '{path}': {exc}") from exc 

262 

263 

264def uninstall_plugin(name: str, registry: RegistryProtocol) -> bool: 

265 """Removes a plugin's directory and deregisters it. 

266 

267 Args: 

268 name (str): The name of the plugin to uninstall. 

269 registry (RegistryProtocol): The plugin registry service. 

270 

271 Returns: 

272 bool: True if the plugin was found and removed, otherwise False. 

273 """ 

274 plug_dir = get_plugins_dir() / name 

275 tel = _tel() 

276 

277 existed = registry.has(name) 

278 if not existed: 

279 if tel: 

280 tel.event("plugin_uninstall_not_found", {"name": name}) 

281 return False 

282 

283 with suppress(Exception): 

284 shutil.rmtree(plug_dir, ignore_errors=True) 

285 

286 registry.deregister(name) 

287 if tel: 

288 tel.event("plugin_uninstalled", {"name": name}) 

289 return True 

290 

291 

292def install_plugin(name: str, force: bool = False, **kwargs: Any) -> None: 

293 """Installs a plugin, handling local paths or pip packages. 

294 

295 If name is a local path, copies to plugins dir. If not, installs via pip. 

296 

297 Args: 

298 name (str): Plugin name or local path. 

299 force (bool): Overwrite if exists. 

300 **kwargs: Additional args (ignored for now). 

301 

302 Raises: 

303 BijuxError: On install failure. 

304 """ 

305 plug_dir = get_plugins_dir() / Path(name).name 

306 if Path(name).exists(): 

307 if plug_dir.exists() and force: 

308 shutil.rmtree(plug_dir) 

309 shutil.copytree(name, plug_dir) 

310 logger.debug(f"Installed local plugin: {name}") 

311 else: 

312 try: 

313 command = [sys.executable, "-m", "pip", "install", name] 

314 subprocess.check_call(command) # noqa: S603 # nosec B603 - controlled command list 

315 logger.debug(f"Pip-installed plugin: {name}") 

316 except Exception as exc: 

317 logger.error(f"Failed to install {name}: {exc}") 

318 raise BijuxError(f"Failed to pip install '{name}': {exc}") from exc 

319 

320 

321def load_entrypoints(registry: RegistryProtocol | None = None) -> list[str]: 

322 """Returns discovered plugin entry point names without importing code.""" 

323 from bijux_cli.plugins.metadata import discover_plugins 

324 

325 names = [meta.name for meta in discover_plugins(strict=False)] 

326 if registry: 

327 logger.debug("Registry provided; discovery is lazy and not registered here") 

328 return names 

329 

330 

331def command_group(*args: Any, **kwargs: Any) -> Any: 

332 """Proxy to the registry command group factory.""" 

333 from bijux_cli.plugins.registry import command_group as _command_group 

334 

335 return _command_group(*args, **kwargs) 

336 

337 

338def dynamic_choices(*args: Any, **kwargs: Any) -> Any: 

339 """Proxy to the registry dynamic choices helper.""" 

340 from bijux_cli.plugins.registry import dynamic_choices as _dynamic_choices 

341 

342 return _dynamic_choices(*args, **kwargs) 

343 

344 

345__all__ = [ 

346 "get_plugins_dir", 

347 "load_plugin_config", 

348 "verify_plugin_signature", 

349 "load_plugin", 

350 "uninstall_plugin", 

351 "install_plugin", 

352 "load_entrypoints", 

353 "command_group", 

354 "dynamic_choices", 

355]