Merge remote-tracking branch 'origin/master' into feat/certified-cross-tenant-worker-pool

docs/architecture/certified-plugins.md

@@ -31,6 +31,19 @@ keys, not private/signing keys. An invalid key-ring configuration is rejected
 rather than weakening verification. Keep active issuer keys during a rotation
 until archives signed by retired IDs are no longer installed.
 
+The ring may also be supplied out-of-band, which is how hosted deployments
+provision it:
+
+```bash
+PLUGIN__CERTIFICATION__TRUSTED_PUBLIC_KEYS_JSON='{"ed25519:issuer":"<base64>"}'
+```
+
+An **empty** ring is a supported state, not a misconfiguration. OSS defaults to
+it, so a self-hosted instance that has not provisioned any issuer key still
+installs packages (see the admission matrix below). Configure the ring to grant
+the shared-runtime profile; leave it empty to keep every package on the
+dedicated profile.
+
 ## Admission matrix
 
 | Deployment | SDK verification | Explicit `administrator_force` | Result |
@@ -40,8 +53,25 @@ until archives signed by retired IDs are no longer installed.
 | Cloud | malformed, untrusted, invalid, or non-shared | any | reject before storage with `CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID` |
 | OSS | absent legacy envelope | any | admitted to the dedicated profile |
 | OSS | valid envelope declaring `shared-runtime-v1` | any | selected shared profile |
-| OSS | malformed or invalid declaration | false | reject with `CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED` |
-| OSS | malformed or invalid declaration | true | admitted to the dedicated profile |
+| OSS | declaration signed by a **key this instance resolves** | false | reject with `CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED` |
+| OSS | declaration signed by a **key this instance resolves** | true | admitted to the dedicated profile |
+| OSS | declaration this instance **cannot resolve** (empty ring) | any | admitted to the dedicated profile |
+
+The OSS row that matters for availability is the last one. Marketplace
+packages are signed by the marketplace issuer and declare
+`shared-runtime-v1`, while OSS ships an empty key ring by default. Treating that
+as a rejection made every certified marketplace package uninstallable with
+`CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED` before artifact storage. Because the
+certificate is signed by an issuer the instance does not declare trusted, no
+shared-runtime privilege may be granted, so admission degrades the install to
+the existing `oss_dev` dedicated profile and records
+`CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED`. This is not an escalation: it
+withholds the shared profile rather than granting it.
+
+A declaration is "resolvable" only when its `key_id` is present in the
+configured ring. When the ring is configured and the declaration still fails
+(malformed, `signature_invalid`, `digest_mismatch`, `unsupported_schema`, ...),
+admission stays explicit and requires `administrator_force`.
 
 `administrator_force` is deliberately strict: it is recognized only when the
 install request carries boolean `true`. The local upload endpoint accepts the

src/langbot/pkg/pipeline/legacy_config_migration.py

@@ -634,6 +634,12 @@ def plan_legacy_pipeline(config, extensions_preferences=None) -> dict:
         return _block(result, 'mixed_runner_selection', 'ai.runner')
     if 'id' in selection:
         current = selection['id']
+        # A blank runner id paired with no legacy runner section means the saved
+        # pipeline simply has no runner selected. There is nothing to migrate and
+        # nothing to convert, so report not_legacy instead of failing the whole
+        # batch with a malformed-id blocker.
+        if type(current) is str and not current.strip() and not any(legacy in ai for legacy in _TARGETS):
+            return result
         if type(current) is not str or not re.fullmatch(r'plugin:[^/\s]+/[^/\s]+/[^/\s]+', current):
             return _block(result, 'invalid_runner_id', 'ai.runner.id')
         result['state'] = 'already_current'

src/langbot/pkg/plugin/certification.py

@@ -48,6 +48,7 @@ class AdmissionCode(str, Enum):
     OSS_FORCE_REQUIRED = 'CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED'
     OSS_FORCED_DEDICATED = 'CERTIFIED_PLUGIN_OSS_FORCED_DEDICATED'
     OSS_CERTIFIED_DEDICATED = 'CERTIFIED_PLUGIN_OSS_CERTIFIED_DEDICATED'
+    OSS_UNTRUSTED_DEDICATED = 'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED'
 
 
 class PluginLogVisibility(str, Enum):
@@ -148,10 +149,17 @@ def verify_plugin_archive_certificate(
 ) -> VerifiedArchiveCertificate:
     """Use the SDK ZIP-comment API and retain its normalized-digest binding."""
 
-    verification = verify_archive(archive, trusted_public_key_ring(trusted_public_keys).get)
+    key_ring = trusted_public_key_ring(trusted_public_keys)
+    verification = verify_archive(archive, key_ring.get)
     envelope = verification.envelope
     runtime_profile = envelope.shared_runtime if envelope is not None else None
-    certificate_id = envelope.key_id if envelope is not None else None
+    # Record the issuer identity only when this instance actually resolved it
+    # through the configured ring. When the ring is empty (for example a
+    # self-hosted deployment that never configured
+    # ``plugin.certification.trusted_public_keys``) the declaration is
+    # unresolvable rather than rejected, so ``certificate_id`` stays unset and
+    # admission can degrade to the dedicated profile instead of blocking.
+    certificate_id = envelope.key_id if envelope is not None and envelope.key_id in key_ring else None
     state = {
         'absent': CertificateVerification.ABSENT,
         'malformed': CertificateVerification.MALFORMED,
@@ -213,6 +221,23 @@ def decide_plugin_admission(
             DEDICATED_RUNTIME,
         )
 
+    # A self-hosted deployment that has not configured the issuer key ring cannot
+    # verify a marketplace archive's declaration. Falling back to the dedicated
+    # runtime (rather than blocking the install) is not a privilege escalation:
+    # the certificate is signed by an issuer this instance does not declare
+    # trusted, so no shared-runtime privilege may be granted. Shared-runtime
+    # isolation is therefore refused while the existing OSS dedicated profile
+    # keeps the install working. Cloud remains fail-closed above.
+    if not trusted_issuer_configured(facts):
+        return PluginAdmissionDecision(
+            AdmissionDisposition.DEDICATED_ALLOWED,
+            AdmissionCode.OSS_UNTRUSTED_DEDICATED,
+            DEDICATED_RUNTIME,
+        )
+
+    # The key ring is configured, yet the declaration still failed to verify
+    # (malformed, signature mismatch, unsupported schema, ...). Surface that as
+    # an explicit decision instead of silently degrading.
     if administrator_force:
         return PluginAdmissionDecision(
             AdmissionDisposition.DEDICATED_ALLOWED,
@@ -227,6 +252,19 @@ def decide_plugin_admission(
     )
 
 
+def trusted_issuer_configured(facts: PluginCertificationFacts) -> bool:
+    """Report whether the deployment holds a trusted issuer key for this archive.
+
+    ``CertificateFacts`` records the verifying key only when it was resolved
+    through the configured ring, so an unresolvable declaration (``unknown_key``)
+    leaves ``certificate_id`` unset. That distinguishes "this operator never
+    configured the issuer" from "a configured trust decision rejected the
+    archive".
+    """
+
+    return bool((facts.certificate.certificate_id or '').strip())
+
+
 def decide_plugin_log_visibility(facts: PluginCertificationFacts) -> PluginLogVisibility:
     """Select the minimum log visibility compatible with a verified shared runtime."""
 

tests/integration/plugin/test_certified_plugin_admission.py

@@ -31,7 +31,7 @@ pytestmark = pytest.mark.integration
         ('cloud', 'signed_shared', False, 'shared-runtime-v1'),
         ('oss', 'signed_shared', False, 'shared-runtime-v1'),
         ('oss', 'legacy', False, 'dedicated'),
-        ('oss', 'invalid_shared', True, 'dedicated'),
+        ('oss', 'forged_shared', True, 'dedicated'),
     ],
 )
 async def test_install_plugin_admits_archive_before_persistence_and_applies_selected_profile(
@@ -90,7 +90,10 @@ async def test_cloud_rejects_untrusted_archive_before_storage_persistence_or_run
 
 @pytest.mark.asyncio
 async def test_oss_requires_explicit_administrator_force_for_declared_invalid_archive() -> None:
-    package, trusted_public_keys = _archive('invalid_shared')
+    # The archive declares the profile of a key this instance *does* trust but is
+    # signed by a different key, so the ring resolves its key_id and the failing
+    # signature is an explicit trust decision rather than an unconfigured ring.
+    package, trusted_public_keys = _archive('forged_shared')
     connector, _execution_context, _binding = _connector('oss', trusted_public_keys)
 
     with pytest.raises(ValueError, match='CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED'):
@@ -101,6 +104,51 @@ async def test_oss_requires_explicit_administrator_force_for_declared_invalid_ar
     connector.handler.apply_plugin_installation.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_oss_admits_unresolvable_declaration_on_the_dedicated_profile() -> None:
+    """A self-hosted instance without the issuer key ring must still install.
+
+    Certified marketplace packages declare ``shared-runtime-v1`` and are signed by
+    the marketplace issuer. An OSS instance that never configured
+    ``plugin.certification.trusted_public_keys`` cannot resolve that issuer, so it
+    must degrade the install to the dedicated profile instead of rejecting every
+    certified package with ``CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED``.
+    """
+
+    package, _trusted_public_keys = _archive('signed_shared')
+    connector, execution_context, binding = _connector('oss', {})
+
+    await connector.install_plugin(PluginInstallSource.LOCAL, {'plugin_file': package})
+
+    persisted_info = connector._persist_installation_package.await_args.kwargs['install_info']
+    assert persisted_info['_certification']['runtime_profile'] == 'dedicated'
+    assert persisted_info['_certification']['admission_code'] == 'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED'
+    assert persisted_info['_certification']['verification'] == 'invalid'
+    connector._store_artifact_package.assert_awaited_once_with(
+        execution_context,
+        hashlib.sha256(package).hexdigest(),
+        package,
+    )
+    connector.handler.apply_plugin_installation.assert_awaited_once_with(
+        binding,
+        artifact_package=package,
+        enabled=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_cloud_rejects_unresolvable_declaration_before_storage() -> None:
+    package, _trusted_public_keys = _archive('signed_shared')
+    connector, _execution_context, _binding = _connector('cloud', {})
+
+    with pytest.raises(ValueError, match='CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'):
+        await connector.install_plugin(PluginInstallSource.LOCAL, {'plugin_file': package})
+
+    connector._store_artifact_package.assert_not_awaited()
+    connector._persist_installation_package.assert_not_awaited()
+    connector.handler.apply_plugin_installation.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('requested_version', [None, '1.0.0'])
 @pytest.mark.parametrize('archive_kind', ['signed_shared', 'legacy'])
@@ -213,6 +261,24 @@ def _archive(kind: str) -> tuple[bytes, dict[str, str]]:
 
     from langbot_plugin.certification import create_envelope, write_envelope
 
+    def _raw_public_key(private_key: Ed25519PrivateKey) -> str:
+        return base64.b64encode(
+            private_key.public_key().public_bytes(
+                serialization.Encoding.Raw,
+                serialization.PublicFormat.Raw,
+            )
+        ).decode('ascii')
+
+    if kind == 'forged_shared':
+        # Declares a key ID the instance trusts but signs with a different key,
+        # so the configured ring resolves the identity and rejects the signature.
+        trusted_key = Ed25519PrivateKey.generate()
+        forged_archive = write_envelope(
+            raw_archive,
+            create_envelope(raw_archive, 'trusted', Ed25519PrivateKey.generate().sign),
+        )
+        return forged_archive, {'trusted': _raw_public_key(trusted_key)}
+
     signing_key = Ed25519PrivateKey.generate()
     signed_archive = write_envelope(
         raw_archive,
@@ -222,11 +288,7 @@ def _archive(kind: str) -> tuple[bytes, dict[str, str]]:
             signing_key.sign,
         ),
     )
-    trusted_key = signing_key.public_key().public_bytes(
-        serialization.Encoding.Raw,
-        serialization.PublicFormat.Raw,
-    )
-    return signed_archive, {'ephemeral': base64.b64encode(trusted_key).decode('ascii')}
+    return signed_archive, {'ephemeral': _raw_public_key(signing_key)}
 
 
 def _normalized_digest(archive: bytes) -> str:

tests/unit_tests/pipeline/test_legacy_config_migration.py

@@ -641,12 +641,40 @@ def test_dify_saved_timeout_is_not_blindly_activated(timeout):
     assert {'code': 'dify.timeout_default', 'field': 'ai.dify-service-api.timeout'} in result['warnings']
 
 
-@pytest.mark.parametrize('current', [None, '', False, [], 'plugin:bad', ' plugin:a/b/c', 'plugin:a/b/c/extra'])
+@pytest.mark.parametrize('current', [None, False, [], 'plugin:bad', ' plugin:a/b/c', 'plugin:a/b/c/extra'])
 def test_malformed_current_id_is_not_already_current(current):
     source = {'ai': {'runner': {'id': current}}}
     assert_block(plan(source), 'invalid_runner_id', 'ai.runner.id')
 
 
+@pytest.mark.parametrize('blank', ['', '   '])
+def test_blank_current_id_without_legacy_section_is_not_legacy(blank):
+    """A saved pipeline that simply has no runner selected is not legacy.
+
+    It has no legacy runner section to convert and no target to synthesize, so it
+    must report not_legacy instead of blocking the whole batch with a
+    malformed-id diagnostic.
+    """
+    source = {'ai': {'runner': {'id': blank, 'expire-time': 0}, 'runner_config': {}}}
+    result = plan(source)
+    assert result['state'] == 'not_legacy'
+    assert result['config'] is None
+    assert result['blockers'] == []
+    assert result['changed_paths'] == []
+    assert result['target_plugin'] is None
+
+
+def test_blank_current_id_with_legacy_section_stays_blocked():
+    """A blank id cannot silently coexist with a legacy section.
+
+    An unselected plugin runner plus a legacy section is an ambiguous state that
+    the operator must resolve, so it keeps the malformed-id blocker.
+    """
+    source = source_for()
+    source['ai']['runner'] = {'id': '', 'expire-time': 0}
+    assert_block(plan(source), 'invalid_runner_id', 'ai.runner.id')
+
+
 @pytest.mark.parametrize('runner', TARGETS)
 def test_deterministic_results_and_input_nonmutation_for_all_nine(runner):
     source = source_for(runner)

tests/unit_tests/plugin/test_certified_plugin_policy.py

@@ -10,27 +10,63 @@ from langbot_plugin.entities.io.context import PluginExecutionMode
 
 
 @pytest.mark.parametrize(
-    ('deployment', 'certificate', 'force', 'expected_disposition', 'expected_code'),
+    ('deployment', 'certificate', 'certificate_id', 'force', 'expected_disposition', 'expected_code'),
     [
-        ('cloud', ('valid', 'shared-runtime-v1'), False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
-        ('cloud', ('absent', None), False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_REQUIRED'),
-        ('cloud', ('malformed', None), False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'),
-        ('cloud', ('invalid', 'shared-runtime-v1'), True, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'),
-        ('oss', ('absent', None), False, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_LEGACY_DEDICATED'),
-        ('oss', ('valid', 'shared-runtime-v1'), False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
+        ('cloud', ('valid', 'shared-runtime-v1'), 'issuer', False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
+        ('cloud', ('absent', None), None, False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_REQUIRED'),
+        ('cloud', ('malformed', None), None, False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'),
+        (
+            'cloud',
+            ('invalid', 'shared-runtime-v1'),
+            'issuer',
+            True,
+            'rejected',
+            'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID',
+        ),
+        ('oss', ('absent', None), None, False, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_LEGACY_DEDICATED'),
+        ('oss', ('valid', 'shared-runtime-v1'), 'issuer', False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
         (
             'oss',
             ('invalid', 'shared-runtime-v1'),
+            'issuer',
             False,
             'administrator_force_required',
             'CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED',
         ),
-        ('oss', ('invalid', 'shared-runtime-v1'), True, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_FORCED_DEDICATED'),
+        (
+            'oss',
+            ('invalid', 'shared-runtime-v1'),
+            'issuer',
+            True,
+            'dedicated_allowed',
+            'CERTIFIED_PLUGIN_OSS_FORCED_DEDICATED',
+        ),
+        # A declaration the operator cannot resolve (no trusted key ring configured)
+        # must keep the OSS install working on the dedicated profile instead of
+        # blocking every certified marketplace package.
+        (
+            'oss',
+            ('invalid', 'shared-runtime-v1'),
+            None,
+            False,
+            'dedicated_allowed',
+            'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED',
+        ),
+        (
+            'oss',
+            ('invalid', 'shared-runtime-v1'),
+            None,
+            True,
+            'dedicated_allowed',
+            'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED',
+        ),
+        ('oss', ('malformed', None), None, False, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED'),
     ],
 )
 def test_admission_policy_enforces_certification_matrix(
     deployment: str,
     certificate: tuple[str, str | None],
+    certificate_id: str | None,
     force: bool,
     expected_disposition: str,
     expected_code: str,
@@ -49,6 +85,7 @@ def test_admission_policy_enforces_certification_matrix(
         certificate=CertificateFacts(
             verification=CertificateVerification(verification),
             runtime_profile=runtime_profile,
+            certificate_id=certificate_id,
         ),
     )