Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 23 additions & 7 deletions anchor/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -804,17 +804,33 @@ def check(ctx, policy, paths, dir, model, metadata, context, server_mode, genera
from anchor.core.sandbox import DiamondCage, install_diamond_cage
cage = DiamondCage()

# 0. ENSURE SANDBOX IS READY (Automatic)
# 0. ENSURE SANDBOX IS READY (Interactive Consent and Verification)
ephemeral_sandbox = False
if not no_sandbox:
if not cage.is_installed():
if verbose: click.secho("Diamond Cage (WASM Sandbox) not found. Initializing...", fg="cyan")
success = install_diamond_cage(verbose=verbose)
if not success:
if verbose: click.secho("WARNING: Sandbox initialization failed. Falling back to host execution.", fg="yellow")
confirm_download = False
if sys.stdin and hasattr(sys.stdin, "isatty") and sys.stdin.isatty():
click.echo("Sandbox runtime not found.")
click.echo("Anchor uses WASMEdge to execute policies inside an isolated sandbox.")
click.echo("The runtime must be downloaded once before sandbox execution.")
confirm_download = click.confirm("Download and install now?", default=True)

if confirm_download:
if verbose: click.secho("Diamond Cage (WASM Sandbox) not found. Initializing...", fg="cyan")
success = install_diamond_cage(verbose=verbose)
if not success:
click.secho("WARNING: Sandbox initialization failed. Falling back to static analysis (no behavioral verification).", fg="yellow")
no_sandbox = True
else:
ephemeral_sandbox = True
if verbose: click.secho("Sandbox Ready.", fg="green")
else:
ephemeral_sandbox = True
if verbose: click.secho("Sandbox Ready.", fg="green")
click.secho("Sandbox installation declined. Disabling sandbox-based evaluation.", fg="yellow")
click.echo("The following capabilities will be unavailable:")
click.echo(" - Safe execution of untrusted code")
click.echo(" - Diamond Cage behavioral sandboxing for agent tools")
click.echo(" - Differential behavioral verification (snapshots)")
no_sandbox = True
else:
if verbose:
click.secho("Using Diamond Cage (WASM Sandbox) for isolation.", fg="cyan")
Expand Down
36 changes: 36 additions & 0 deletions anchor/core/sandbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,11 +188,20 @@ class DiamondCage:
"windows_x86_64":f"https://github.com/WasmEdge/WasmEdge/releases/download/{WASMEDGE_VERSION}/WasmEdge-{WASMEDGE_VERSION}-windows.zip",
}

WASMEDGE_SHA256 = {
"linux_x86_64": "3686e0226871bf17b62ec57e1c15778c2947834b90af0dfad14f2e0202bf9284",
"darwin_x86_64": "b7fdfaf59805951241f47690917b501ddfa06d9b6f7e0262e44e784efe4a7b33",
"darwin_arm64": "acc93721210294ced0887352f360e42e46dcc05332e6dd78c1452fb3a35d5255",
"windows_x86_64":"db533289ba26ec557b5193593c9ed03db75be3bc7aa737e2caa5b56b8eef888a",
}

PYTHON_WASM_URL = (
"https://github.com/vmware-labs/webassembly-language-runtimes/releases/"
"download/python%2F3.11.3%2B20230428-7d1b259/python-3.11.3.wasm"
)

PYTHON_WASM_SHA256 = "658cb6add2bbf8dfe84d67fb85430956d5a8b2b1d694d33432d654cdf048f813"

def __init__(self, anchor_home: Optional[Path] = None, verbose: bool = False):
self.anchor_home = anchor_home or (Path.home() / ".anchor")
self.verbose = verbose
Expand Down Expand Up @@ -619,6 +628,7 @@ def install_diamond_cage(force: bool = False, verbose: bool = False) -> bool:
import urllib.request
import tarfile
import zipfile
import hashlib

cage = DiamondCage(verbose=verbose)

Expand Down Expand Up @@ -655,6 +665,20 @@ def install_diamond_cage(force: bool = False, verbose: bool = False) -> bool:
with open(archive_path, "wb") as f:
f.write(response.read())

# Verify WasmEdge archive integrity
expected_wasmedge_hash = cage.WASMEDGE_SHA256.get(platform_key)
if expected_wasmedge_hash:
if verbose:
print(" Verifying WasmEdge archive integrity...")
hasher = hashlib.sha256()
with open(archive_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
actual_hash = hasher.hexdigest()
if actual_hash != expected_wasmedge_hash:
archive_path.unlink()
raise ValueError(f"WasmEdge archive integrity verification failed! Hash mismatch.\nExpected: {expected_wasmedge_hash}\nActual: {actual_hash}")

if verbose:
print("PKP: Extracting WasmEdge...")
if cage.os_name == "windows":
Expand Down Expand Up @@ -720,6 +744,18 @@ def install_diamond_cage(force: bool = False, verbose: bool = False) -> bool:
with open(cage.python_wasm_path, "wb") as f:
f.write(response.read())

# Verify Python WASM integrity
if verbose:
print(" Verifying Python WASM binary integrity...")
hasher = hashlib.sha256()
with open(cage.python_wasm_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
actual_hash = hasher.hexdigest()
if actual_hash != cage.PYTHON_WASM_SHA256:
cage.python_wasm_path.unlink()
raise ValueError(f"Python WASM integrity verification failed! Hash mismatch.\nExpected: {cage.PYTHON_WASM_SHA256}\nActual: {actual_hash}")

if verbose:
print("[V] Python WASM installed.")

Expand Down
81 changes: 81 additions & 0 deletions tests/unit/test_sandbox_consent.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
import os
import sys
from unittest.mock import patch, MagicMock
from click.testing import CliRunner
from anchor.cli import check

def test_sandbox_consent_declined():
runner = CliRunner()

with patch("anchor.cli.sys") as mock_sys, \
patch("anchor.core.sandbox.DiamondCage") as mock_cage_class, \
patch("click.confirm") as mock_confirm:

# Simulate sandbox not installed
mock_cage = MagicMock()
mock_cage.is_installed.return_value = False
mock_cage_class.return_value = mock_cage

# Simulate interactive TTY
mock_sys.stdin = MagicMock()
mock_sys.stdin.isatty.return_value = True

# User declines download
mock_confirm.return_value = False

# Invoke check command
result = runner.invoke(check, [".", "--no-sign"])

# Verify confirm prompt was displayed and clicked
mock_confirm.assert_called_with("Download and install now?", default=True)
assert "Sandbox installation declined. Disabling sandbox-based evaluation." in result.output

def test_sandbox_consent_accepted_success():
runner = CliRunner()

with patch("anchor.cli.sys") as mock_sys, \
patch("anchor.core.sandbox.DiamondCage") as mock_cage_class, \
patch("anchor.core.sandbox.install_diamond_cage") as mock_install, \
patch("click.confirm") as mock_confirm:

mock_cage = MagicMock()
mock_cage.is_installed.return_value = False
mock_cage_class.return_value = mock_cage

mock_sys.stdin = MagicMock()
mock_sys.stdin.isatty.return_value = True

# User accepts download
mock_confirm.return_value = True
# Installation succeeds
mock_install.return_value = True

result = runner.invoke(check, [".", "--no-sign"])

mock_confirm.assert_called_with("Download and install now?", default=True)
mock_install.assert_called_once()

def test_sandbox_integrity_verification_failure():
# Test that a hash mismatch aborts installation in install_diamond_cage
from anchor.core.sandbox import install_diamond_cage

with patch("os.path.exists") as mock_exists, \
patch("urllib.request.urlopen") as mock_urlopen, \
patch("hashlib.sha256") as mock_sha256:

mock_exists.return_value = True

# Mock download response data
mock_response = MagicMock()
mock_response.read.return_value = b"corrupted data"
mock_urlopen.return_value.__enter__.return_value = mock_response

# Mock hasher to return a bad/non-matching hash
mock_hasher = MagicMock()
mock_hasher.hexdigest.return_value = "bad_hash_value"
mock_sha256.return_value = mock_hasher

# Run install - should return False because WasmEdge verification fails
# (with verbose=True to print errors)
success = install_diamond_cage(force=True, verbose=True)
assert success is False
Loading