orchestrata_ultimate.plugin.governance

Reference › orchestrata_ultimate › plugin › governance

PluginGovernance holds the admission and protection rules for plugins. It is pure Python: every check compares against data in the PluginConfig, with no Prolog engine involved. The PluginManager creates it and calls is_plugin_allowed before every load.

is_plugin_allowed(manifest) refuses any blocklisted name; in explicit mode it admits only allowlisted names, and in open mode every other plugin. is_capability_granted grants a capability only if the manifest declares it. can_register_volk and can_register_tool refuse reserved core Volk aliases and reserved system tool names, and log a warning; they return a boolean and do not raise. get_resource_limit(resource) maps a ResourceType (memory bytes, API calls per hour, events per hour) to the default from the configured limits and raises ValueError for any other value. The capability names are defined in interfaces.

At a glance

Class diagram

Dependency diagram

API

Classes

PluginGovernance

Line 42

Governance enforcement for plugin operations.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/plugin/governance.py

"""Plugin permission governance.

Provides pure-Python allowlist/blocklist and reserved-name checks for plugin
loading, capability grants, resource limits, and registration authorization.
This is not Prolog-backed: there is no Prolog engine or rule base involved
anywhere in this module -- every check here is a plain Python comparison
against ``PluginConfig`` data (allowlist/blocklist/reserved-name sets).
"""

import logging

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.config import PluginConfig
from orchestrata_ultimate.plugin.interfaces import ResourceType
from orchestrata_ultimate.plugin.manifest import PluginManifest

_LOG = logging.getLogger(__name__)

_RESERVED_VOLK_ALIASES = frozenset(
    {
        "U:Ox",
        "U:PS",
        "U:KR",
        "U:Vz",
        "U:Ux",
        "U:PM",
        "U:Ar",
        "U:Qa",
        "U:Cx",
    }
)

_RESERVED_TOOL_NAMES = frozenset(
    {
        "system.shell",
        "system.eval",
        "system.exec",
    }
)


class PluginGovernance:
    """Governance enforcement for plugin operations.

    Checks plugin allowlist/blocklist, capability declarations,
    reserved aliases, and resource limits. Pure Python: every
    check is a direct comparison against ``PluginConfig`` data --
    there is no Prolog engine or rule base involved.

    Attributes:
        config: Plugin manager configuration.
    """

    def __init__(self, config: PluginConfig) -> None:
        """Initialize governance with plugin config.

        Args:
            config: Plugin manager configuration with
                allowlist_mode, allowlist, blocklist.
        """
        self._config = config

    @traces(
        spec="015",
        story="US-009",
        reqs=["REQ-015-114"],
    )
    def is_plugin_allowed(self, manifest: PluginManifest) -> bool:
        """Check if a plugin is allowed to load.

        In "explicit" mode, only allowlisted plugins are
        permitted. In "open" mode, any non-blocklisted plugin
        is permitted.

        Args:
            manifest: Plugin manifest to check.

        Returns:
            True if the plugin is allowed to load.
        """
        name = manifest.name
        if name in self._config.blocklist:
            _LOG.info("Plugin '%s' is blocklisted", name)
            return False
        if self._config.allowlist_mode == "explicit":
            allowed = name in self._config.allowlist
            if not allowed:
                _LOG.info(
                    "Plugin '%s' not in allowlist (mode=explicit)",
                    name,
                )
            return allowed
        return True

    def is_capability_granted(
        self,
        plugin_name: str,
        capability: str,
        declared_capabilities: list[str],
    ) -> bool:
        """Check if a capability is granted to a plugin.

        A capability is granted only if the plugin declares it
        in its manifest and the plugin itself is allowed.

        Args:
            plugin_name: Plugin name.
            capability: Capability string to check.
            declared_capabilities: Capabilities from manifest.

        Returns:
            True if the capability is granted.
        """
        granted = capability in declared_capabilities
        if not granted:
            _LOG.info(
                "Capability '%s' not granted to plugin '%s' (not declared)",
                capability,
                plugin_name,
            )
        return granted

    @traces(
        spec="015",
        story="US-002",
        reqs=["REQ-015-020"],
    )
    def can_register_volk(self, plugin_name: str, alias: str) -> bool:
        """Check if a plugin can register a Volk alias.

        Reserved core Volk aliases cannot be overridden.

        Args:
            plugin_name: Plugin name.
            alias: Proposed Volk alias.

        Returns:
            True if the alias is not reserved.
        """
        if alias in _RESERVED_VOLK_ALIASES:
            _LOG.warning(
                "Plugin '%s' tried to register reserved alias '%s'",
                plugin_name,
                alias,
            )
            return False
        return True

    def can_register_tool(self, plugin_name: str, tool_name: str) -> bool:
        """Check if a plugin can register a tool name.

        Reserved system tool names cannot be overridden.

        Args:
            plugin_name: Plugin name.
            tool_name: Proposed tool name.

        Returns:
            True if the tool name is not reserved.
        """
        if tool_name in _RESERVED_TOOL_NAMES:
            _LOG.warning(
                "Plugin '%s' tried to register reserved tool '%s'",
                plugin_name,
                tool_name,
            )
            return False
        return True

    @traces(
        spec="015",
        story="US-009",
        reqs=["REQ-015-119"],
    )
    def get_resource_limit(self, resource: ResourceType) -> int:
        """Get the default resource limit for the given type.

        Returns the default limit from PluginConfig for the
        specified resource type.

        Args:
            resource: Resource type enum member.

        Returns:
            The limit value as an integer.

        Raises:
            ValueError: If resource is not a valid ResourceType.
        """
        if not isinstance(resource, ResourceType):
            raise ValueError(
                f"Unknown resource type: {resource!r}. "
                f"Use ResourceType enum members: "
                f"{[r.value for r in ResourceType]}",
            )
        limits = self._config.resource_limits
        if resource == ResourceType.MEMORY_BYTES:
            return limits.default_memory_bytes
        if resource == ResourceType.API_CALLS_PER_HOUR:
            return limits.default_api_calls_per_hour
        return limits.default_events_per_hour