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

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 discovered = discover_plugin_manifests() 

39 ordered = resolve_plugin_dependencies( 

40 discovered, 

41 registered_plugins=target.registered_plugins(), 

42 ) 

43 for plugin in ordered: 

44 if require_signatures: 

45 signing_info = signature_verifier(plugin) 

46 else: 

47 signing_info = allow_unsigned_development_override(plugin) 

48 logger.warning( 

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

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

51 plugin.manifest.name, 

52 ) 

53 signing_identity = signing_info.signing_identity 

54 signing_metadata = signing_info.audit_metadata() 

55 target.register_plugin( 

56 plugin.manifest, 

57 discovery_source=_discovery_source(plugin), 

58 signing_identity=signing_identity, 

59 signing_metadata=signing_metadata, 

60 ) 

61 if freeze: 

62 target.freeze() 

63 if ordered: 

64 logger.info( 

65 "registered installed plugins: %s", 

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

67 ) 

68 return ordered 

69 

70 

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

72 return { 

73 "type": "python_entry_point", 

74 "entry_point_name": plugin.entry_point_name, 

75 "entry_point_value": plugin.entry_point_value, 

76 "distribution": plugin.distribution, 

77 }