capabilities

capabilities

RBAC capability system — fine-grained permission model for tool dispatch.

Classes

Capability

Bases: str, Enum

Fine-grained capability labels.

CapabilityGrant dataclass

CapabilityGrant(capability: str, pattern: str = '*')

A single capability grant for an agent.

AgentPolicy dataclass

AgentPolicy(agent_id: str, grants: List[CapabilityGrant] = list(), deny: List[str] = list())

Policy for a specific agent.

CapabilityPolicy

CapabilityPolicy(*, policy_path: Optional[str] = None, default_deny: bool = False)

RBAC capability policy for tool dispatch.

Checks whether an agent has the required capability to invoke a tool. Policy can be loaded from a JSON file or configured programmatically.

Default policy: if no explicit policy exists for an agent, all capabilities are granted (open by default). Set default_deny=True to flip to deny-by-default.

Source code in src/openjarvis/security/capabilities.py

def __init__(
    self,
    *,
    policy_path: Optional[str] = None,
    default_deny: bool = False,
) -> None:
    self._policies: Dict[str, AgentPolicy] = {}
    self._default_deny = default_deny

    from openjarvis._rust_bridge import get_rust_module

    _rust = get_rust_module()
    self._rust_impl = _rust.CapabilityPolicy(default_deny=default_deny)

    if policy_path:
        self._load_file(Path(policy_path))

Functions

grant

grant(agent_id: str, capability: str, pattern: str = '*') -> None

Grant a capability to an agent.

Source code in src/openjarvis/security/capabilities.py

def grant(self, agent_id: str, capability: str, pattern: str = "*") -> None:
    """Grant a capability to an agent."""
    policy = self._policies.setdefault(
        agent_id,
        AgentPolicy(agent_id=agent_id),
    )
    policy.grants.append(CapabilityGrant(capability=capability, pattern=pattern))
    self._rust_impl.grant(agent_id, capability, pattern)

deny

deny(agent_id: str, capability: str) -> None

Explicitly deny a capability to an agent.

Source code in src/openjarvis/security/capabilities.py

def deny(self, agent_id: str, capability: str) -> None:
    """Explicitly deny a capability to an agent."""
    policy = self._policies.setdefault(
        agent_id,
        AgentPolicy(agent_id=agent_id),
    )
    policy.deny.append(capability)
    self._rust_impl.deny(agent_id, capability)

check

check(agent_id: str, capability: str, resource: str = '') -> bool

Check whether agent_id has capability for resource.

Falls back to the _default wildcard agent's grants only when agent_id has no explicit policy of its own. This lets a baseline be granted once via grant("_default", ...) instead of needing to know every dynamically-created managed-agent UUID ahead of time while preserving the invariant that an agent-specific denial always wins.

Source code in src/openjarvis/security/capabilities.py

def check(self, agent_id: str, capability: str, resource: str = "") -> bool:
    """Check whether *agent_id* has *capability* for *resource*.

    Falls back to the ``_default`` wildcard agent's grants only when
    *agent_id* has no explicit policy of its own. This lets a baseline be
    granted once via
    ``grant("_default", ...)`` instead of needing to know every
    dynamically-created managed-agent UUID ahead of time while preserving
    the invariant that an agent-specific denial always wins.
    """
    if (
        not agent_id
        or agent_id == self._DEFAULT_AGENT
        or agent_id in self._policies
    ):
        return self._rust_impl.check(agent_id, capability, resource)
    if self._DEFAULT_AGENT in self._policies:
        return self._rust_impl.check(self._DEFAULT_AGENT, capability, resource)
    return self._rust_impl.check(agent_id, capability, resource)

list_grants

list_grants(agent_id: str) -> List[CapabilityGrant]

List all grants for an agent.

Source code in src/openjarvis/security/capabilities.py

def list_grants(self, agent_id: str) -> List[CapabilityGrant]:
    """List all grants for an agent."""
    policy = self._policies.get(agent_id)
    return list(policy.grants) if policy else []

list_agents

list_agents() -> List[str]

List all agents with explicit policies.

Source code in src/openjarvis/security/capabilities.py

def list_agents(self) -> List[str]:
    """List all agents with explicit policies."""
    return list(self._policies.keys())

save

save(path: Path) -> None

Save policy to a JSON file.

Source code in src/openjarvis/security/capabilities.py

def save(self, path: Path) -> None:
    """Save policy to a JSON file."""
    agents = []
    for agent_id, policy in self._policies.items():
        agents.append(
            {
                "agent_id": agent_id,
                "grants": [
                    {"capability": g.capability, "pattern": g.pattern}
                    for g in policy.grants
                ],
                "deny": policy.deny,
            }
        )
    path.write_text(json.dumps({"agents": agents}, indent=2))

Functions

canonical_tool_capabilities

canonical_tool_capabilities(tool: Any) -> List[str]

Return the non-bypassable capability floor for tool.

Third-party tools remain governed by their ToolSpec. An in-tree tool that was newly registered without being inventoried fails closed as system:admin instead of silently becoming unrestricted.

Source code in src/openjarvis/security/capabilities.py

def canonical_tool_capabilities(tool: Any) -> List[str]:
    """Return the non-bypassable capability floor for *tool*.

    Third-party tools remain governed by their ToolSpec.  An in-tree tool that
    was newly registered without being inventoried fails closed as
    ``system:admin`` instead of silently becoming unrestricted.
    """
    module = type(tool).__module__
    name = tool.spec.name
    is_builtin = (
        module == "openjarvis.tools"
        or module.startswith("openjarvis.tools.")
        or module == "openjarvis.scheduler.tools"
    )
    if module == "openjarvis.tools.mcp_adapter":
        # MCP tool names are remote-controlled.  Resolve adapter provenance
        # before the name table so a server cannot impersonate a reviewed-safe
        # local tool such as ``calculator`` or ``think``.
        return [Capability.TOOL_INVOKE]
    if is_builtin:
        if name in DEFAULT_TOOL_CAPABILITIES:
            canonical = list(DEFAULT_TOOL_CAPABILITIES[name])
            expected = _SAFE_BUILTIN_PROVENANCE.get(name)
            if (
                expected is not None
                and (
                    module,
                    type(tool).__name__,
                )
                != expected
            ):
                logger.error(
                    "Tool %r claimed reviewed-safe built-in provenance from %s.%s",
                    name,
                    module,
                    type(tool).__name__,
                )
                return [Capability.SYSTEM_ADMIN]
            return canonical
        logger.error("Built-in tool %r has no canonical capability inventory", name)
        return [Capability.SYSTEM_ADMIN]
    if name in DEFAULT_TOOL_CAPABILITIES:
        canonical = list(DEFAULT_TOOL_CAPABILITIES[name])
        # Third-party tools that collide with a privileged name retain its
        # security floor.  Reviewed-safe names are safe only for their in-tree
        # implementation and therefore fail closed on foreign provenance.
        return canonical or [Capability.SYSTEM_ADMIN]
    return []