Coverage for node / src / stigmem_node / plugins / lifecycle.py: 84%

1"""Installed plugin registration lifecycle.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6 

7from stigmem_node.settings import settings 

8 

9from .discovery import DiscoveredPlugin, discover_plugin_manifests, resolve_plugin_dependencies 

10from .registry import HookRegistry, get_registry 

11from .signing import ( 

12 PluginSignatureVerifier, 

13 allow_unsigned_development_override, 

14 require_verified_signature, 

15) 

16 

17logger = logging.getLogger("stigmem.plugins.lifecycle") 

18 

19 

20def register_discovered_plugins( 

21 *, 

22 registry: HookRegistry | None = None, 

23 freeze: bool = True, 

24 signing_required: bool | None = None, 

25 signature_verifier: PluginSignatureVerifier = require_verified_signature, 

26) -> tuple[DiscoveredPlugin, ...]: 

27 """Discover, dependency-order, and register installed plugin packages. 

28 

29 This function is intentionally startup-only. It mutates the supplied registry 

30 during application initialization and freezes it by default so later runtime 

31 hot unload/reload paths are not introduced by PR 4-INF.2. 

32 """ 

33 

34 target = registry or get_registry() 

35 require_signatures = ( 

36 settings.plugin_signing_required if signing_required is None else signing_required 

37 ) 

38 if not settings.plugin_auto_discovery_enabled: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 logger.info("installed plugin auto-discovery disabled") 

40 if freeze: 

41 target.freeze() 

42 return () 

43 

44 discovered = discover_plugin_manifests() 

45 ordered = resolve_plugin_dependencies( 

46 discovered, 

47 registered_plugins=target.registered_plugins(), 

48 ) 

49 for plugin in ordered: 

50 if require_signatures: 

51 signing_info = signature_verifier(plugin) 

52 else: 

53 signing_info = allow_unsigned_development_override(plugin) 

54 logger.warning( 

55 "SECURITY WARNING: unsigned plugin %r loaded because " 

56 "STIGMEM_PLUGIN_SIGNING_REQUIRED=false; do not use this setting in production", 

57 plugin.manifest.name, 

58 ) 

59 signing_identity = signing_info.signing_identity 

60 signing_metadata = signing_info.audit_metadata() 

61 target.register_plugin( 

62 plugin.manifest, 

63 discovery_source=_discovery_source(plugin), 

64 signing_identity=signing_identity, 

65 signing_metadata=signing_metadata, 

66 ) 

67 if freeze: 

68 target.freeze() 

69 if ordered: 

70 logger.info( 

71 "registered installed plugins: %s", 

72 ", ".join(plugin.manifest.name for plugin in ordered), 

73 ) 

74 return ordered 

75 

76 

77def _discovery_source(plugin: DiscoveredPlugin) -> dict[str, str | None]: 

78 return { 

79 "type": "python_entry_point", 

80 "entry_point_name": plugin.entry_point_name, 

81 "entry_point_value": plugin.entry_point_value, 

82 "distribution": plugin.distribution, 

83 }