Darshan Poudel (rsd-darshan) · Jun 1, 2026 · Jun 1, 2026 · Jun 1, 2026 · Jun 2, 2026 · Jun 2, 2026
@@ -32,6 +32,16 @@ pip install "agt-sandbox[hyperlight]"
 pip install "agt-sandbox[azure,policy]"
 ```
 
+> **Ring enforcement note:** `agt-sandbox[full]` does not include the
+> `[hypervisor]` extra. To enable `SandboxConfig.ring`-based enforcement
+> (hypervisor ring constraints applied at session creation and
+> `execute_code` time), install with:
+> ```bash
+> pip install "agt-sandbox[full,hypervisor]"
+> ```
+> Without this extra, `SandboxConfig.ring` is accepted but the ring
+> enforcement code path is never entered.
+
 The Azure data-plane SDK ships as an early-access wheel — pin the URL:
 
 ```bash

@@ -53,6 +53,13 @@ azure = [
 policy = [
     "agent-governance-toolkit-core>=4.0.0,<5.0",
 ]
+# Ring enforcement requires the hypervisor package. Kept optional so the
+# core agent-sandbox install does not depend on a package version that may
+# not yet be published. When SandboxConfig.ring is None the ring-enforcement
+# code path is never entered and this extra is not needed.
+hypervisor = [
+    "agent_hypervisor>=3.7.0,<5.0",
+]
 full = [
     "agt-sandbox[docker,hyperlight,policy]",
 ]

@@ -118,6 +118,7 @@ def aca_config_from_policy(
         input_dir=base.input_dir,
         output_dir=base.output_dir,
         runtime=base.runtime,
+        ring=base.ring,
     )
 
     defaults = getattr(policy, "defaults", None)
@@ -560,6 +561,22 @@ def create_session(
                     f"Failed to initialize PolicyEvaluator: {exc}"
                 ) from exc
 
+        # Apply ring constraints — only when a ring is explicitly set.
+        if cfg.ring is not None:
+            from hypervisor.rings.enforcer import RING_CONSTRAINTS
+            ring_constraints = RING_CONSTRAINTS[cfg.ring]
+            if not ring_constraints.network_allowed:
+                if allow_hosts:
+                    logger.info(
+                        "Ring %s: clearing network_allowlist for agent '%s' "
+                        "(network not permitted at this ring)",
+                        cfg.ring.value,
+                        agent_id,
+                    )
+                allow_hosts = []
+                net_default = "deny"
+                policy_provided = True  # ensure _apply_egress_policy is called below
+
         # Make sure the sandbox group exists (no-op unless requested).
         self._ensure_sandbox_group()
 

@@ -273,6 +273,7 @@ def docker_config_from_policy(
         input_dir=base.input_dir,
         output_dir=base.output_dir,
         runtime=base.runtime,
+        ring=base.ring,
     )
 
     # Resource limits from policy defaults
@@ -553,6 +554,50 @@ def create_session(
                     f"Failed to initialize PolicyEvaluator: {exc}"
                 ) from exc
 
+        # 1b. Apply ring constraints — only when a ring is explicitly set.
+        # ring=None means the hypervisor extra is not installed; skip silently.
+        if cfg.ring is not None:
+            from hypervisor.rings.enforcer import RING_CONSTRAINTS
+            ring_constraints = RING_CONSTRAINTS[cfg.ring]
+            if not ring_constraints.network_allowed and cfg.network_enabled:
+                logger.info(
+                    "Ring %s: overriding network_enabled=False for agent '%s'",
+                    cfg.ring.value,
+                    agent_id,
+                )
+                cfg = SandboxConfig(
+                    timeout_seconds=cfg.timeout_seconds,
+                    memory_mb=cfg.memory_mb,
+                    cpu_limit=cfg.cpu_limit,
+                    network_enabled=False,
+                    read_only_fs=cfg.read_only_fs,
+                    env_vars=cfg.env_vars,
+                    input_dir=cfg.input_dir,
+                    output_dir=cfg.output_dir,
+                    runtime=cfg.runtime,
+                    output_max_bytes=cfg.output_max_bytes,
+                    ring=cfg.ring,
+                )
+            if ring_constraints.filesystem_scope == "none" and not cfg.read_only_fs:
+                logger.info(
+                    "Ring %s: overriding read_only_fs=True for agent '%s'",
+                    cfg.ring.value,
+                    agent_id,
+                )
+                cfg = SandboxConfig(
+                    timeout_seconds=cfg.timeout_seconds,
+                    memory_mb=cfg.memory_mb,
+                    cpu_limit=cfg.cpu_limit,
+                    network_enabled=cfg.network_enabled,
+                    read_only_fs=True,
+                    env_vars=cfg.env_vars,
+                    input_dir=cfg.input_dir,
+                    output_dir=cfg.output_dir,
+                    runtime=cfg.runtime,
+                    output_max_bytes=cfg.output_max_bytes,
+                    ring=cfg.ring,
+                )
+
         # 2. Create hardened container
         container = self._create_container(agent_id, session_id, cfg)
         with self._state_lock:
@@ -600,6 +645,16 @@ def execute_code(
                     f"Policy denied: {decision.reason}"
                 )
 
+        # Ring resource check — only when a ring is explicitly set.
+        if session_cfg is not None and session_cfg.ring is not None:
+            from hypervisor.rings.enforcer import ResourceType, RingEnforcer
+            ring_result = RingEnforcer().check_resource(session_cfg.ring, ResourceType.SUBPROCESS)
+            if not ring_result.allowed:
+                raise PermissionError(
+                    f"Ring {session_cfg.ring.value} agent cannot execute "
+                    f"subprocess: {ring_result.reason}"
+                )
+
         enforce_no_subprocess_execution(code)
 
         # Run code with the session's configured timeout/env, not defaults.

@@ -287,6 +287,7 @@ def __init__(
         self._sandboxes: dict[tuple[str, str], Any] = {}
         self._evaluators: dict[tuple[str, str], Any] = {}
         self._session_configs: dict[tuple[str, str], HyperlightConfig] = {}
+        self._session_rings: dict[tuple[str, str], Any] = {}
         self._snapshots: dict[tuple[str, str, str], Any] = {}
 
         # Resolve the upstream SDK lazily but eagerly enough to set
@@ -384,6 +385,30 @@ def create_session(
             net_allow = list(getattr(policy, "network_allowlist", []) or [])
             evaluator = self._build_evaluator(policy)
 
+        # Apply ring constraints before resolving tool and network capabilities.
+        # Only active when a ring is explicitly set; ring=None skips silently.
+        if base_cfg.ring is not None:
+            from hypervisor.rings.enforcer import RING_CONSTRAINTS
+            ring_constraints = RING_CONSTRAINTS[base_cfg.ring]
+            if not ring_constraints.subprocess_allowed:
+                if tool_allow:
+                    logger.info(
+                        "Ring %s: clearing tool_allowlist for agent '%s' "
+                        "(subprocess not permitted at this ring)",
+                        base_cfg.ring.value,
+                        agent_id,
+                    )
+                tool_allow = []
+            if not ring_constraints.network_allowed:
+                if net_allow:
+                    logger.info(
+                        "Ring %s: clearing network_allowlist for agent '%s' "
+                        "(network not permitted at this ring)",
+                        base_cfg.ring.value,
+                        agent_id,
+                    )
+                net_allow = []
+
         # Resolve tool callables. Names listed in the allowlist that the
         # provider does not know about fail closed at session creation
         # time so a misconfigured policy never silently degrades.
@@ -441,6 +466,7 @@ def _bootstrap_sandbox() -> Any:
             self._workers[(agent_id, session_id)] = worker
             self._sandboxes[(agent_id, session_id)] = sandbox
             self._session_configs[(agent_id, session_id)] = hl_cfg
+            self._session_rings[(agent_id, session_id)] = base_cfg.ring
             if evaluator is not None:
                 self._evaluators[(agent_id, session_id)] = evaluator
 
@@ -474,6 +500,7 @@ def execute_code(
             sandbox = self._sandboxes.get(key)
             evaluator = self._evaluators.get(key)
             cfg = self._session_configs.get(key)
+            session_ring = self._session_rings.get(key)
 
         if worker is None or sandbox is None:
             raise RuntimeError(
@@ -496,6 +523,16 @@ def execute_code(
                 reason = getattr(decision, "reason", "policy denied")
                 raise PermissionError(f"Policy denied: {reason}")
 
+        # Ring resource check — only when a ring is explicitly set.
+        if session_ring is not None:
+            from hypervisor.rings.enforcer import ResourceType, RingEnforcer
+            ring_result = RingEnforcer().check_resource(session_ring, ResourceType.SUBPROCESS)
+            if not ring_result.allowed:
+                raise PermissionError(
+                    f"Ring {session_ring.value} agent cannot execute "
+                    f"subprocess: {ring_result.reason}"
+                )
+
         enforce_no_subprocess_execution(code)
 
         execution_id = uuid.uuid4().hex[:8]
@@ -554,6 +591,7 @@ def destroy_session(self, agent_id: str, session_id: str) -> None:
             sandbox = self._sandboxes.pop(key, None)
             self._evaluators.pop(key, None)
             self._session_configs.pop(key, None)
+            self._session_rings.pop(key, None)
             # Pop any snapshots associated with this session into a
             # separate list so we can drop them on the worker thread
             # below — snapshot objects share the unsendable invariant

@@ -24,7 +24,10 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any
+from typing import TYPE_CHECKING, Any, Optional
+
+if TYPE_CHECKING:
+    from hypervisor.models import ExecutionRing
 
 logger = logging.getLogger(__name__)
 
@@ -66,6 +69,11 @@ class SandboxConfig:
 
     Extends the minimal config with fields needed by session-based
     providers (``input_dir``, ``output_dir``, ``runtime``).
+
+    The ``ring`` field maps the agent's execution ring to resource
+    constraints enforced at session-creation time.  Defaults to
+    ``RING_3_SANDBOX`` (most restrictive) so sandboxes are fail-closed
+    unless the caller explicitly grants a higher ring.
     """
 
     timeout_seconds: float = 60.0
@@ -78,6 +86,7 @@ class SandboxConfig:
     output_dir: str | None = None
     runtime: str | None = None
     output_max_bytes: int = 1_048_576  # 1 MiB per stream
+    ring: Optional[Any] = None  # ExecutionRing; None is treated as RING_3_SANDBOX at enforcement time
 
 
 @dataclass

@@ -773,3 +773,133 @@ def test_top_level_exports_hyperlight(self):
         assert agent_sandbox.HyperlightConfig is not None
         assert agent_sandbox.SnapshotHandle is not None
         assert "HyperLightSandboxProvider" in agent_sandbox.__all__
+
+
+# =========================================================================
+# 11. Ring enforcement
+# =========================================================================
+
+
+class _FakeConstraints:
+    def __init__(self, *, network_allowed: bool, subprocess_allowed: bool) -> None:
+        self.network_allowed = network_allowed
+        self.subprocess_allowed = subprocess_allowed
+
+
+class _FakeRingCheckResult:
+    def __init__(self, *, allowed: bool, reason: str = "") -> None:
+        self.allowed = allowed
+        self.reason = reason
+
+
+class _FakeRingEnforcer:
+    def check_resource(self, ring: Any, resource_type: Any) -> _FakeRingCheckResult:
+        if getattr(ring, "value", None) == 3:
+            return _FakeRingCheckResult(
+                allowed=False, reason="ring 3 disallows subprocess"
+            )
+        return _FakeRingCheckResult(allowed=True)
+
+
+def _install_fake_hypervisor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> tuple[Any, Any]:
+    """Stub hypervisor.models and hypervisor.rings.enforcer in sys.modules.
+
+    Returns (ring_2, ring_3) objects that can be used as SandboxConfig.ring.
+    """
+    ring_2 = SimpleNamespace(value=2, name="RING_2_STANDARD")
+    ring_3 = SimpleNamespace(value=3, name="RING_3_SANDBOX")
+
+    constraints = {
+        ring_2: _FakeConstraints(network_allowed=True, subprocess_allowed=True),
+        ring_3: _FakeConstraints(network_allowed=False, subprocess_allowed=False),
+    }
+
+    models_mod = types.ModuleType("hypervisor.models")
+    models_mod.ExecutionRing = SimpleNamespace(  # type: ignore[attr-defined]
+        RING_2_STANDARD=ring_2,
+        RING_3_SANDBOX=ring_3,
+    )
+
+    enforcer_mod = types.ModuleType("hypervisor.rings.enforcer")
+    enforcer_mod.RING_CONSTRAINTS = constraints  # type: ignore[attr-defined]
+    enforcer_mod.RingEnforcer = _FakeRingEnforcer  # type: ignore[attr-defined]
+    enforcer_mod.ResourceType = SimpleNamespace(  # type: ignore[attr-defined]
+        SUBPROCESS="subprocess",
+    )
+
+    monkeypatch.setitem(sys.modules, "hypervisor", types.ModuleType("hypervisor"))
+    monkeypatch.setitem(
+        sys.modules, "hypervisor.rings", types.ModuleType("hypervisor.rings")
+    )
+    monkeypatch.setitem(sys.modules, "hypervisor.models", models_mod)
+    monkeypatch.setitem(sys.modules, "hypervisor.rings.enforcer", enforcer_mod)
+
+    return ring_2, ring_3
+
+
+class TestRingEnforcement:
+    """Ring-level constraint enforcement in HyperLightSandboxProvider."""
+
+    def test_ring3_clears_network_allowlist(self, fake_sdk, monkeypatch):
+        """RING_3 with a non-empty network_allowlist gets all domains cleared."""
+        _ring2, ring3 = _install_fake_hypervisor(monkeypatch)
+        from agent_sandbox.hyperlight_provider import HyperLightSandboxProvider
+
+        provider = HyperLightSandboxProvider()
+        policy = _make_policy(
+            network_allowlist=["api.example.com", "cdn.example.com"]
+        )
+        cfg = SandboxConfig(ring=ring3)
+        handle = provider.create_session("agent-r3", policy=policy, config=cfg)
+
+        sandbox = _FakeSandbox.instances[-1]
+        assert sandbox.allowed_domains == [], (
+            "RING_3 must clear the network allowlist; "
+            f"got {sandbox.allowed_domains}"
+        )
+
+    def test_ring2_preserves_network_allowlist(self, fake_sdk, monkeypatch):
+        """RING_2 leaves the network_allowlist intact."""
+        ring2, _ring3 = _install_fake_hypervisor(monkeypatch)
+        from agent_sandbox.hyperlight_provider import HyperLightSandboxProvider
+
+        provider = HyperLightSandboxProvider()
+        policy = _make_policy(network_allowlist=["api.example.com"])
+        cfg = SandboxConfig(ring=ring2)
+        handle = provider.create_session("agent-r2", policy=policy, config=cfg)
+
+        sandbox = _FakeSandbox.instances[-1]
+        assert "api.example.com" in sandbox.allowed_domains, (
+            "RING_2 must preserve the network allowlist; "
+            f"got {sandbox.allowed_domains}"
+        )
+
+    def test_ring3_execute_raises_permission_error(self, fake_sdk, monkeypatch):
+        """execute_code on a RING_3 session raises PermissionError."""
+        _ring2, ring3 = _install_fake_hypervisor(monkeypatch)
+        from agent_sandbox.hyperlight_provider import HyperLightSandboxProvider
+
+        provider = HyperLightSandboxProvider()
+        cfg = SandboxConfig(ring=ring3)
+        handle = provider.create_session("agent-r3-exec", config=cfg)
+
+        with pytest.raises(PermissionError, match="subprocess"):
+            provider.execute_code(
+                handle.agent_id, handle.session_id, "print('hello')"
+            )
+
+    def test_no_ring_no_enforcement(self, fake_sdk, monkeypatch):
+        """SandboxConfig with ring=None skips enforcement entirely."""
+        _ring2, ring3 = _install_fake_hypervisor(monkeypatch)
+        from agent_sandbox.hyperlight_provider import HyperLightSandboxProvider
+
+        provider = HyperLightSandboxProvider()
+        policy = _make_policy(network_allowlist=["api.example.com"])
+        cfg = SandboxConfig(ring=None)
+        handle = provider.create_session("agent-no-ring", policy=policy, config=cfg)
+
+        sandbox = _FakeSandbox.instances[-1]
+        # No ring enforcement → allowlist preserved as-is
+        assert "api.example.com" in sandbox.allowed_domains