fix(plugin): load certification key ring from env (#2554)

src/langbot/pkg/core/stages/load_config.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 import copy
+import json
 from typing import Any
 from langbot.pkg.utils import bounded_executor, constants
 import yaml
@@ -215,6 +216,30 @@ def _apply_env_overrides_to_config(cfg: dict) -> dict:
     return cfg
 
 
+def _apply_certification_key_ring_env(cfg: dict) -> dict:
+    """Load the public certification key ring from one strict JSON env value.
+
+    The generic environment override intentionally skips dictionaries. This
+    narrow exception keeps trusted issuer keys deployable without relying on a
+    mutable persisted config file, while rejecting malformed input instead of
+    silently running with an empty trust ring.
+    """
+    raw = os.getenv('PLUGIN__CERTIFICATION__TRUSTED_PUBLIC_KEYS_JSON')
+    if raw is None:
+        return cfg
+    try:
+        key_ring = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise ValueError('PLUGIN__CERTIFICATION__TRUSTED_PUBLIC_KEYS_JSON must be valid JSON') from exc
+    if not isinstance(key_ring, dict) or any(
+        not isinstance(key_id, str) or not key_id.strip() or not isinstance(key, str) or not key.strip()
+        for key_id, key in key_ring.items()
+    ):
+        raise ValueError('PLUGIN__CERTIFICATION__TRUSTED_PUBLIC_KEYS_JSON must be a non-empty string-to-string mapping')
+    cfg['plugin']['certification']['trusted_public_keys'] = key_ring
+    return cfg
+
+
 @stage.stage_class('LoadConfigStage')
 class LoadConfigStage(stage.BootingStage):
     """Load config file stage"""
@@ -268,6 +293,7 @@ class LoadConfigStage(stage.BootingStage):
 
         # Apply environment variable overrides to data/config.yaml
         ap.instance_config.data = _apply_env_overrides_to_config(ap.instance_config.data)
+        ap.instance_config.data = _apply_certification_key_ring_env(ap.instance_config.data)
 
         blocking_config = ap.instance_config.data['system']['blocking_executor']
         ap.blocking_executor = bounded_executor.configure_bounded_default_executor(

tests/unit_tests/core/test_load_config.py

@@ -427,3 +427,27 @@ class TestApplyEnvOverridesToConfig:
             result = load_config._apply_env_overrides_to_config(cfg)
 
         assert result['api']['extra_webhook_prefix'] == 'https://extra.example.com'
+
+
+class TestCertificationKeyRingEnv:
+    def test_applies_string_key_mapping_from_strict_json(self):
+        load_config = get_load_config_module()
+        cfg = load_config._complete_runtime_policy_defaults({})
+        with patch.dict(
+            os.environ,
+            {'PLUGIN__CERTIFICATION__TRUSTED_PUBLIC_KEYS_JSON': '{"ed25519:issuer":"YWJj"}'},
+            clear=True,
+        ):
+            result = load_config._apply_certification_key_ring_env(cfg)
+        assert result['plugin']['certification']['trusted_public_keys'] == {'ed25519:issuer': 'YWJj'}
+
+    def test_rejects_malformed_or_non_mapping_key_ring(self):
+        load_config = get_load_config_module()
+        for value in ('not-json', '[]', '{"":"YWJj"}', '{"ed25519:issuer": 1}'):
+            cfg = load_config._complete_runtime_policy_defaults({})
+            with patch.dict(os.environ, {'PLUGIN__CERTIFICATION__TRUSTED_PUBLIC_KEYS_JSON': value}, clear=True):
+                try:
+                    load_config._apply_certification_key_ring_env(cfg)
+                except ValueError:
+                    continue
+            raise AssertionError(f'invalid certification key ring was accepted: {value!r}')