Merge remote-tracking branch 'origin/master' into feat/workspace-operation-traceability

src/langbot/pkg/persistence/alembic/versions/0032_certification_artifact_digest_backfill.py

@@ -0,0 +1,138 @@
+"""Backfill raw artifact digests for legacy shared certificates.
+
+Revision ID: 0032_cert_artifact_digest
+Revises: 0031_merge_totp_assistant
+
+Older certified-plugin rows persisted every shared-admission fact except the raw
+archive digest.  Placement now requires that digest to match the installation's
+immutable artifact digest, so this migration fills only records whose remaining
+facts already prove the exact legacy shared-admission shape.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '0032_cert_artifact_digest'
+down_revision = '0031_merge_totp_assistant'
+branch_labels = None
+depends_on = None
+
+_TABLE = 'plugin_settings'
+_SHARED_RUNTIME = 'shared-runtime-v1'
+_SHARED_ADMISSION_CODE = 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'
+_LOWER_HEX = frozenset('0123456789abcdef')
+_HEX = frozenset('0123456789abcdefABCDEF')
+
+
+def _is_digest(value: object, *, lowercase: bool) -> bool:
+    allowed = _LOWER_HEX if lowercase else _HEX
+    return isinstance(value, str) and len(value) == 64 and all(character in allowed for character in value)
+
+
+def _eligible_certification(install_info: object, artifact_digest: object) -> Mapping[object, object] | None:
+    if not _is_digest(artifact_digest, lowercase=True) or not isinstance(install_info, Mapping):
+        return None
+    certification = install_info.get('_certification')
+    if not isinstance(certification, Mapping) or 'artifact_digest' in certification:
+        return None
+    certificate_id = certification.get('certificate_id')
+    if not isinstance(certificate_id, str) or not certificate_id.strip():
+        return None
+    if not _is_digest(certification.get('normalized_digest'), lowercase=False):
+        return None
+    required_facts = {
+        'verification': 'valid',
+        'certificate_runtime_profile': _SHARED_RUNTIME,
+        'runtime_profile': _SHARED_RUNTIME,
+        'admission_code': _SHARED_ADMISSION_CODE,
+    }
+    if not all(certification.get(key) == value for key, value in required_facts.items()):
+        return None
+    return certification
+
+
+def _suspend_postgres_rls(conn: sa.Connection) -> tuple[bool, bool]:
+    if conn.dialect.name != 'postgresql':
+        return False, False
+    state = conn.execute(
+        sa.text('SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE oid = to_regclass(:table_name)'),
+        {'table_name': _TABLE},
+    ).one()
+    rls_enabled, rls_forced = bool(state.relrowsecurity), bool(state.relforcerowsecurity)
+    if rls_forced:
+        conn.execute(sa.text(f'ALTER TABLE {_TABLE} NO FORCE ROW LEVEL SECURITY'))
+    if rls_enabled:
+        conn.execute(sa.text(f'ALTER TABLE {_TABLE} DISABLE ROW LEVEL SECURITY'))
+    return rls_enabled, rls_forced
+
+
+def _restore_postgres_rls(conn: sa.Connection, state: tuple[bool, bool]) -> None:
+    if conn.dialect.name != 'postgresql':
+        return
+    rls_enabled, rls_forced = state
+    if rls_enabled:
+        conn.execute(sa.text(f'ALTER TABLE {_TABLE} ENABLE ROW LEVEL SECURITY'))
+    if rls_forced:
+        conn.execute(sa.text(f'ALTER TABLE {_TABLE} FORCE ROW LEVEL SECURITY'))
+
+
+def backfill_certification_artifact_digests(conn: sa.Connection) -> None:
+    inspector = sa.inspect(conn)
+    if _TABLE not in inspector.get_table_names():
+        return
+    columns = {column['name'] for column in inspector.get_columns(_TABLE)}
+    required_columns = {
+        'workspace_uuid',
+        'plugin_author',
+        'plugin_name',
+        'artifact_digest',
+        'install_info',
+    }
+    if not required_columns <= columns:
+        return
+
+    plugin_settings = sa.table(
+        _TABLE,
+        sa.column('workspace_uuid', sa.String(36)),
+        sa.column('plugin_author', sa.String(255)),
+        sa.column('plugin_name', sa.String(255)),
+        sa.column('artifact_digest', sa.String(64)),
+        sa.column('install_info', sa.JSON()),
+    )
+    rows = conn.execute(sa.select(plugin_settings)).mappings().all()
+    for row in rows:
+        certification = _eligible_certification(row['install_info'], row['artifact_digest'])
+        if certification is None:
+            continue
+        updated_certification = dict(certification)
+        updated_certification['artifact_digest'] = row['artifact_digest']
+        updated_install_info = dict(row['install_info'])
+        updated_install_info['_certification'] = updated_certification
+        conn.execute(
+            plugin_settings.update()
+            .where(plugin_settings.c.workspace_uuid == row['workspace_uuid'])
+            .where(plugin_settings.c.plugin_author == row['plugin_author'])
+            .where(plugin_settings.c.plugin_name == row['plugin_name'])
+            .values(install_info=updated_install_info)
+        )
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if _TABLE not in sa.inspect(conn).get_table_names():
+        return
+    rls_state = _suspend_postgres_rls(conn)
+    try:
+        backfill_certification_artifact_digests(conn)
+    finally:
+        _restore_postgres_rls(conn, rls_state)
+
+
+def downgrade() -> None:
+    # The source digest cannot be distinguished from a digest persisted by
+    # current Core, so downgrade intentionally preserves the safe enrichment.
+    pass

src/langbot/pkg/plugin/certification.py

@@ -14,6 +14,7 @@ from enum import Enum
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 from langbot_plugin.certification import normalized_zip_digest, verify_archive
+from langbot_plugin.entities.io.context import PluginExecutionMode
 
 
 SHARED_RUNTIME_V1 = 'shared-runtime-v1'
@@ -270,3 +271,37 @@ def decide_plugin_log_visibility(facts: PluginCertificationFacts) -> PluginLogVi
     if facts.certificate.is_valid_shared_runtime:
         return PluginLogVisibility.TENANT_SCOPED
     return PluginLogVisibility.DETAILED_PROCESS
+
+
+def execution_mode_for_persisted_installation(
+    *,
+    artifact_digest: str,
+    install_info: object,
+) -> PluginExecutionMode:
+    """Derive placement only from persisted facts bound to the exact artifact."""
+
+    if len(artifact_digest) != 64 or any(character not in '0123456789abcdef' for character in artifact_digest.lower()):
+        return PluginExecutionMode.DEDICATED
+    if not isinstance(install_info, Mapping):
+        return PluginExecutionMode.DEDICATED
+    certification = install_info.get('_certification')
+    if not isinstance(certification, Mapping):
+        return PluginExecutionMode.DEDICATED
+    normalized_digest = certification.get('normalized_digest')
+    if not isinstance(normalized_digest, str) or len(normalized_digest) != 64:
+        return PluginExecutionMode.DEDICATED
+    if any(character not in '0123456789abcdef' for character in normalized_digest.lower()):
+        return PluginExecutionMode.DEDICATED
+    certificate_id = certification.get('certificate_id')
+    if not isinstance(certificate_id, str) or not certificate_id.strip():
+        return PluginExecutionMode.DEDICATED
+    shared_facts = {
+        'artifact_digest': artifact_digest,
+        'verification': CertificateVerification.VALID.value,
+        'certificate_runtime_profile': SHARED_RUNTIME_V1,
+        'runtime_profile': SHARED_RUNTIME_V1,
+        'admission_code': AdmissionCode.SHARED_ELIGIBLE.value,
+    }
+    if all(certification.get(key) == value for key, value in shared_facts.items()):
+        return PluginExecutionMode.SHARED_CERTIFIED
+    return PluginExecutionMode.DEDICATED

src/langbot/pkg/plugin/connector.py

@@ -34,6 +34,7 @@ from .certification import (
     PluginCertificationFacts,
     VerifiedArchiveCertificate,
     decide_plugin_admission,
+    execution_mode_for_persisted_installation,
     verify_plugin_archive_certificate,
 )
 from .github import (
@@ -62,6 +63,7 @@ from langbot_plugin.runtime.security import (
 )
 from langbot_plugin.entities.io.context import (
     InstallationBinding,
+    PluginExecutionMode,
     PluginInstallationDesiredState,
     PluginWorkerPolicy,
     RuntimeIdentity,
@@ -343,6 +345,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             artifact_digest=setting.artifact_digest,
         )
 
+    @staticmethod
+    def _execution_mode_from_setting(setting: persistence_plugin.PluginSetting) -> PluginExecutionMode:
+        return execution_mode_for_persisted_installation(
+            artifact_digest=setting.artifact_digest,
+            install_info=setting.install_info,
+        )
+
     def _legacy_oss_bridge_binding(self, execution_context: ExecutionContext) -> InstallationBinding:
         seed = f'langbot:oss-plugin-bridge:{execution_context.instance_uuid}:{execution_context.workspace_uuid}'
         return InstallationBinding(
@@ -511,6 +520,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 PluginInstallationDesiredState(
                     binding=binding,
                     enabled=setting.enabled,
+                    execution_mode=self._execution_mode_from_setting(setting),
                 )
             )
         return tuple(desired_states)
@@ -526,6 +536,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             desired.binding,
             artifact_package=artifact_package,
             enabled=desired.enabled,
+            execution_mode=desired.execution_mode,
         )
         self._raise_apply_failure(desired, result)
         if result.get('state') != 'artifact_missing':
@@ -551,6 +562,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             desired.binding,
             artifact_package=persisted_package,
             enabled=desired.enabled,
+            execution_mode=desired.execution_mode,
         )
         self._raise_apply_failure(desired, repaired)
         if repaired.get('state') == 'artifact_missing':
@@ -1793,6 +1805,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         }:
             raise ValueError(decision.code.value)
         certification_info = {
+            'artifact_digest': hashlib.sha256(file_bytes).hexdigest(),
             'normalized_digest': facts.artifact_digest,
             'verification': facts.certificate.verification.value,
             'certificate_runtime_profile': facts.certificate.runtime_profile,
@@ -1893,7 +1906,14 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                     install_info=install_info,
                     artifact_digest=artifact_digest,
                 )
-                desired = PluginInstallationDesiredState(binding=binding, enabled=True)
+                desired = PluginInstallationDesiredState(
+                    binding=binding,
+                    enabled=True,
+                    execution_mode=execution_mode_for_persisted_installation(
+                        artifact_digest=artifact_digest,
+                        install_info=install_info,
+                    ),
+                )
                 runtime_handler.register_installation_binding(
                     binding,
                     plugin_author=plugin_author,
@@ -2143,6 +2163,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         desired = PluginInstallationDesiredState(
             binding=binding,
             enabled=setting.enabled,
+            execution_mode=self._execution_mode_from_setting(setting),
         )
         is_legacy_oss = self.runtime_profile == 'oss_dev' and (
             not isinstance(setting.install_info, dict)

src/langbot/pkg/plugin/handler.py

@@ -24,6 +24,7 @@ from langbot_plugin.entities.io.context import (
     ActionContext,
     ApplyPluginInstallationRequest,
     InstallationBinding,
+    PluginExecutionMode,
     PluginInstallationDesiredState,
     PluginWorkerPolicy,
     ReconcilePluginInstallationsRequest,
@@ -2793,6 +2794,7 @@ class RuntimeConnectionHandler(handler.Handler):
         *,
         artifact_package: bytes | None,
         enabled: bool,
+        execution_mode: PluginExecutionMode = PluginExecutionMode.DEDICATED,
     ) -> dict[str, Any]:
         with self.installation_scope(binding):
             artifact_file_key = None
@@ -2801,6 +2803,7 @@ class RuntimeConnectionHandler(handler.Handler):
             request = ApplyPluginInstallationRequest(
                 artifact_file_key=artifact_file_key,
                 enabled=enabled,
+                execution_mode=execution_mode,
             )
             return await self.call_action(
                 LangBotToRuntimeAction.APPLY_PLUGIN_INSTALLATION,