Files
JWeBot/packages/jwebot-kernel/src/jwebot_kernel/extension_dispatch.py
T

258 lines
9.0 KiB
Python

"""Signed startup catalog and isolated extension-host dispatch."""
from __future__ import annotations
import os
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Any, Protocol
from jwebot_sdk import API_VERSION, ContractViolation, ExtensionResult
from .activation import verify_installed_release_signature
from .catalog import ExtensionCatalog
from .host_client import ServiceHandler
from .manifest import PluginManifest, load_manifest
class HostedExtensionError(ValueError):
"""Controlled signed-catalog, health, or invocation failure."""
@dataclass(frozen=True)
class ExpectedExtension:
plugin_id: str
author: str
commands: tuple[str, ...]
service_profiles: tuple[str, ...]
description: str = ""
help_entries: tuple[str, ...] = ()
@dataclass(frozen=True)
class HostedRelease:
manifest: PluginManifest
release_dir: Path
class HostInvoker(Protocol):
def __call__(
self,
request: Mapping[str, Any],
service_handler: ServiceHandler | None,
) -> Mapping[str, Any]: ...
def _active_release_dir(plugin_root: Path) -> Path | None:
current = plugin_root / "current"
if not current.exists() and not current.is_symlink():
return None
if not current.is_symlink():
raise HostedExtensionError("extension current pointer is not a symlink")
target = Path(os.readlink(current))
if (
target.is_absolute()
or len(target.parts) != 2
or target.parts[0] != "releases"
):
raise HostedExtensionError("extension current pointer is unsafe")
release_dir = plugin_root / target
if not release_dir.is_dir() or release_dir.is_symlink():
raise HostedExtensionError("extension current release is unavailable")
return release_dir
class SignedExtensionInventory:
"""Load reviewed signed manifests during startup, then freeze routing."""
def __init__(
self,
catalog: ExtensionCatalog,
releases: Mapping[str, HostedRelease],
) -> None:
if not catalog.frozen:
raise HostedExtensionError("signed extension catalog must be frozen")
self.catalog = catalog
self._releases = MappingProxyType(dict(releases))
@property
def releases(self) -> Mapping[str, HostedRelease]:
return self._releases
@classmethod
def load(
cls,
state_root: Path,
allowed_signers: Path,
expected: Sequence[ExpectedExtension],
*,
expected_publisher: str | None = None,
require_all: bool = False,
) -> "SignedExtensionInventory":
catalog = ExtensionCatalog()
releases: dict[str, HostedRelease] = {}
missing: list[str] = []
for descriptor in sorted(expected, key=lambda item: item.plugin_id):
plugin_root = state_root / descriptor.plugin_id
release_dir = _active_release_dir(plugin_root)
if release_dir is None:
missing.append(descriptor.plugin_id)
continue
try:
record, _ = verify_installed_release_signature(
release_dir,
allowed_signers,
expected_publisher=expected_publisher,
)
manifest = load_manifest(release_dir / "plugin.toml")
except Exception as error:
raise HostedExtensionError(
f"signed extension verification failed: {descriptor.plugin_id}"
) from error
if (
record.plugin_id != descriptor.plugin_id
or record.author != descriptor.author
or manifest.plugin_id != descriptor.plugin_id
or manifest.author != descriptor.author
or (
descriptor.description
and manifest.description != descriptor.description
)
or (
descriptor.help_entries
and manifest.help_entries != descriptor.help_entries
)
or manifest.capabilities.commands != descriptor.commands
or manifest.capabilities.service_profiles
!= descriptor.service_profiles
):
raise HostedExtensionError(
f"signed extension inventory mismatch: {descriptor.plugin_id}"
)
catalog.register(manifest)
releases[descriptor.plugin_id] = HostedRelease(
manifest=manifest,
release_dir=release_dir,
)
if require_all and missing:
raise HostedExtensionError(
"required signed extensions are missing: "
+ ", ".join(sorted(missing))
)
catalog.freeze()
return cls(catalog, releases)
class HostedExtensionRuntime:
"""Health-gated dispatch over signed, frozen Extension Host releases."""
def __init__(
self,
inventory: SignedExtensionInventory,
invokers: Mapping[str, HostInvoker],
service_handler: ServiceHandler,
*,
require_all_healthy: bool = False,
) -> None:
self.inventory = inventory
self.service_handler = service_handler
active: dict[str, HostInvoker] = {}
failures: dict[str, str] = {}
for plugin_id, release in inventory.releases.items():
invoker = invokers.get(plugin_id)
if invoker is None:
failures[plugin_id] = "HostInvokerMissing"
continue
try:
response = invoker({"op": "health"}, None)
self._validate_health(plugin_id, release.manifest, response)
except Exception as error:
failures[plugin_id] = type(error).__name__
continue
active[plugin_id] = invoker
if require_all_healthy and len(active) != len(inventory.releases):
unavailable = sorted(set(inventory.releases) - set(active))
raise HostedExtensionError(
"required extension hosts are unhealthy: "
+ ", ".join(unavailable)
)
self._active = MappingProxyType(active)
self._failures = MappingProxyType(failures)
@staticmethod
def _validate_health(
plugin_id: str,
manifest: PluginManifest,
response: Mapping[str, Any],
) -> None:
if (
response.get("status") != "ok"
or response.get("api_version") != API_VERSION
or response.get("plugin_id") != plugin_id
or response.get("author") != manifest.author
or response.get("description") != manifest.description
or tuple(response.get("help", ())) != manifest.help_entries
or tuple(response.get("commands", ()))
!= manifest.capabilities.commands
or tuple(response.get("service_profiles", ()))
!= manifest.capabilities.service_profiles
):
raise HostedExtensionError("extension host health response mismatch")
@property
def active_plugin_ids(self) -> frozenset[str]:
return frozenset(self._active)
@property
def failures(self) -> Mapping[str, str]:
return self._failures
def available(self, plugin_id: str) -> bool:
return plugin_id in self._active
def invoke(
self,
plugin_id: str,
command: str,
arguments: str,
event: Mapping[str, Any],
) -> ExtensionResult:
invoker = self._active.get(plugin_id)
if invoker is None:
raise HostedExtensionError("extension host is not active")
release = self.inventory.releases.get(plugin_id)
if release is None:
raise HostedExtensionError("extension release is not registered")
if command.casefold() not in {
value.casefold()
for value in release.manifest.capabilities.commands
}:
raise HostedExtensionError("extension command is not declared")
try:
response = invoker(
{
"op": "invoke_command",
"command": command,
"arguments": arguments,
"event": dict(event),
},
self.service_handler,
)
except Exception as error:
raise HostedExtensionError("extension host invocation failed") from error
if (
response.get("status") != "ok"
or response.get("api_version") != API_VERSION
or response.get("plugin_id") != plugin_id
or not isinstance(response.get("result"), Mapping)
):
raise HostedExtensionError("extension host response mismatch")
try:
return ExtensionResult.from_mapping(response["result"])
except ContractViolation as error:
raise HostedExtensionError(
"extension host result violates the SDK contract"
) from error