725 lines
27 KiB
Python
725 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Mandatory Linux/Nono 0.74 behavioral gate for Slice 1 core policies."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
from contextlib import ExitStack, closing
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from tests.nono.capabilities.core.policy import compose_profile, isolated_settings, load_profile
|
|
from tests.nono.harness.fixtures import isolated_environment
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[4]
|
|
PROFILES = ROOT / "profiles" / "nono"
|
|
LM_STUDIO = ROOT / "extensions" / "lm-studio" / "index.ts"
|
|
PI_CLI = Path(
|
|
"/home/jb/.bun/install/global/node_modules/"
|
|
"@earendil-works/pi-coding-agent/dist/bundle/cli.js"
|
|
)
|
|
PACKAGE_LOCK = ROOT / "config" / "pi-package-lock.json"
|
|
LEGACY_LM_STUDIO = Path("/home/jb/.pi/agent/extensions/lm-studio.ts")
|
|
ACTIVE_STATE_PATHS = (
|
|
Path("/home/jb/.pi/agent/settings.json"),
|
|
LEGACY_LM_STUDIO,
|
|
Path("/home/jb/.config/nono/profiles/pi-local.json"),
|
|
Path("/home/jb/.config/nono/profiles/pi-work.json"),
|
|
Path("/home/jb/.config/nono/profiles/pi-orca-local.json"),
|
|
Path("/home/jb/.config/nono/profiles/pi-harness-tools.json"),
|
|
)
|
|
RAW_CANARIES = {
|
|
"ANTHROPIC_API_KEY": "slice1-anthropic-canary",
|
|
"GH_TOKEN": "slice1-gh-canary",
|
|
"GITHUB_TOKEN": "slice1-github-canary",
|
|
"MORPH_API_KEY": "slice1-morph-canary",
|
|
"OPENAI_API_KEY": "slice1-openai-canary",
|
|
"SSH_AUTH_SOCK": "/tmp/slice1-forbidden-agent.sock",
|
|
}
|
|
|
|
|
|
class HostGateError(RuntimeError):
|
|
"""The mandatory Slice 1 host proof failed."""
|
|
|
|
|
|
def _active_state_snapshot() -> dict[str, tuple[str, str, int]]:
|
|
snapshot: dict[str, tuple[str, str, int]] = {}
|
|
for path in ACTIVE_STATE_PATHS:
|
|
if path.is_symlink():
|
|
target = str(path.readlink())
|
|
payload = path.resolve(strict=True).read_bytes()
|
|
elif path.is_file():
|
|
target = "regular"
|
|
payload = path.read_bytes()
|
|
else:
|
|
target = "absent"
|
|
payload = b""
|
|
snapshot[str(path)] = (
|
|
target,
|
|
hashlib.sha256(payload).hexdigest(),
|
|
path.lstat().st_mode if path.exists() or path.is_symlink() else 0,
|
|
)
|
|
return snapshot
|
|
|
|
|
|
def _read_regular_file(path: Path, *, limit: int = 1024 * 1024) -> bytes:
|
|
descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
|
|
try:
|
|
before = os.fstat(descriptor)
|
|
if not stat.S_ISREG(before.st_mode) or before.st_size > limit:
|
|
raise HostGateError(f"expected one bounded regular file: {path}")
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
while True:
|
|
chunk = os.read(descriptor, min(65536, limit + 1 - total))
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
total += len(chunk)
|
|
if total > limit:
|
|
raise HostGateError(f"file exceeds verification limit: {path}")
|
|
after = os.fstat(descriptor)
|
|
if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != (
|
|
after.st_dev,
|
|
after.st_ino,
|
|
after.st_size,
|
|
after.st_mtime_ns,
|
|
):
|
|
raise HostGateError(f"file changed during verification: {path}")
|
|
return b"".join(chunks)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def _validate_lm_studio_sources() -> bytes:
|
|
lock = json.loads(PACKAGE_LOCK.read_text(encoding="utf-8"))
|
|
entries = [
|
|
entry for entry in lock.get("extensions", []) if entry.get("name") == "lm-studio"
|
|
]
|
|
if len(entries) != 1:
|
|
raise HostGateError("package lock must contain exactly one LM Studio entry")
|
|
reviewed = entries[0].get("reviewed", {})
|
|
managed_source = _read_regular_file(LM_STUDIO)
|
|
managed_digest = hashlib.sha256(managed_source).hexdigest()
|
|
if reviewed.get("sha256") != managed_digest:
|
|
raise HostGateError("managed LM Studio source does not match package-lock evidence")
|
|
legacy_source = _read_regular_file(LEGACY_LM_STUDIO)
|
|
legacy_digest = hashlib.sha256(legacy_source).hexdigest()
|
|
if reviewed.get("legacySourceSha256") != legacy_digest:
|
|
raise HostGateError("legacy LM Studio source drifted before approved quarantine")
|
|
return managed_source
|
|
|
|
|
|
def _validate_static_profiles(nono: str, environment: dict[str, str]) -> None:
|
|
profiles = sorted(PROFILES.glob("pi-core-*.json")) + sorted(
|
|
PROFILES.glob("provider-*.json")
|
|
)
|
|
for profile in profiles:
|
|
result = subprocess.run(
|
|
[nono, "profile", "validate", "--strict", str(profile), "--json"],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
if result.returncode != 0:
|
|
raise HostGateError(f"strict validation failed for {profile}: {result.stdout}")
|
|
|
|
|
|
def _prove_provider_fragment_is_not_launchable(
|
|
nono: str, environment: dict[str, str]
|
|
) -> None:
|
|
result = subprocess.run(
|
|
[
|
|
nono,
|
|
"run",
|
|
"--profile",
|
|
str(PROFILES / "provider-lm-studio.json"),
|
|
"--",
|
|
"/usr/bin/env",
|
|
],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
input="",
|
|
check=False,
|
|
timeout=20,
|
|
)
|
|
if result.returncode == 0 or "directory is not readable" not in (
|
|
result.stdout + result.stderr
|
|
):
|
|
raise HostGateError("provider fragment was directly launchable")
|
|
|
|
|
|
def _prove_nono_composition(nono: str, session: Any, base_environment: dict[str, str]) -> None:
|
|
profiles = session.home / ".config" / "nono" / "profiles"
|
|
profiles.mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
os.chmod(profiles, 0o700)
|
|
(profiles / "slice-one-parent-a.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"extends": "default",
|
|
"meta": {"name": "slice-one-parent-a"},
|
|
"groups": {"exclude": ["system_write_linux", "user_tools"]},
|
|
"network": {"block": True},
|
|
"environment": {
|
|
"allow_vars": ["PATH"],
|
|
"deny_vars": ["SLICE_ONE_SECRET_A"],
|
|
},
|
|
"open_urls": {"allow_origins": [], "allow_localhost": False},
|
|
"allow_launch_services": False,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(profiles / "slice-one-parent-b.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"extends": "default",
|
|
"meta": {"name": "slice-one-parent-b"},
|
|
"network": {"block": False, "connect_port": [1234]},
|
|
"environment": {
|
|
"allow_vars": ["HOME"],
|
|
"deny_vars": ["SLICE_ONE_SECRET_B"],
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(profiles / "slice-one-child.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"extends": ["slice-one-parent-a", "slice-one-parent-b"],
|
|
"meta": {"name": "slice-one-child"},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
for profile_name, parent_name, decision in (
|
|
("slice-one-command-parent", "default", "deny"),
|
|
("slice-one-command-child", "slice-one-command-parent", "allow"),
|
|
):
|
|
(profiles / f"{profile_name}.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"extends": parent_name,
|
|
"meta": {"name": profile_name},
|
|
"groups": {"exclude": ["system_write_linux"]},
|
|
"command_policies": {
|
|
"commands": {
|
|
"echo": {
|
|
"executable": "/usr/bin/echo",
|
|
"from": {
|
|
"session": {
|
|
"sandbox": {
|
|
"fs_read_file": ["/usr/bin/echo"]
|
|
},
|
|
"invocation_policy": {"default": decision},
|
|
}
|
|
},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
environment = dict(base_environment)
|
|
environment.update(
|
|
{
|
|
"SLICE_ONE_SECRET_A": "canary-a",
|
|
"SLICE_ONE_SECRET_B": "canary-b",
|
|
"SLICE_ONE_UNLISTED": "canary-c",
|
|
}
|
|
)
|
|
shown = subprocess.run(
|
|
[nono, "profile", "show", "slice-one-child", "--json"],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
if shown.returncode != 0:
|
|
raise HostGateError(shown.stderr + shown.stdout)
|
|
resolved = json.loads(shown.stdout)
|
|
if resolved.get("network", {}).get("block") is not True:
|
|
raise HostGateError("Nono multiple inheritance widened a blocked network")
|
|
observed = subprocess.run(
|
|
[nono, "run", "--profile", "slice-one-child", "--", "/usr/bin/env"],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
input="",
|
|
check=False,
|
|
timeout=20,
|
|
)
|
|
if observed.returncode != 0:
|
|
raise HostGateError(observed.stderr + observed.stdout)
|
|
child_environment = {
|
|
line.partition("=")[0]
|
|
for line in observed.stdout.splitlines()
|
|
if "=" in line
|
|
}
|
|
if not {"HOME", "PATH"}.issubset(child_environment):
|
|
raise HostGateError("Nono did not merge parent environment allowlists")
|
|
if child_environment.intersection(
|
|
{"SLICE_ONE_SECRET_A", "SLICE_ONE_SECRET_B", "SLICE_ONE_UNLISTED"}
|
|
):
|
|
raise HostGateError("Nono environment composition exposed denied variables")
|
|
command_policy = subprocess.run(
|
|
[
|
|
nono,
|
|
"why",
|
|
"--profile",
|
|
"slice-one-command-child",
|
|
"--command",
|
|
"echo",
|
|
"--json",
|
|
"--",
|
|
"hello",
|
|
],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
if command_policy.returncode != 0:
|
|
raise HostGateError(command_policy.stderr + command_policy.stdout)
|
|
try:
|
|
command_decision = json.loads(command_policy.stdout)
|
|
except json.JSONDecodeError as error:
|
|
raise HostGateError(
|
|
f"Nono command policy query returned invalid JSON: "
|
|
f"{command_policy.stdout!r} {command_policy.stderr!r}"
|
|
) from error
|
|
if command_decision.get("status") != "denied":
|
|
raise HostGateError("Nono child command policy widened an inherited denial")
|
|
if command_decision.get("policy_source") != (
|
|
"command_policies.commands.echo.from.session.invocation_policy"
|
|
):
|
|
raise HostGateError("Nono command policy used an unexpected caller edge")
|
|
|
|
|
|
def _listener(port: int, stack: ExitStack, http_requests: dict[int, int]) -> None:
|
|
server = stack.enter_context(closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)))
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
server.bind(("127.0.0.1", port))
|
|
except OSError as error:
|
|
raise HostGateError(f"loopback fixture port {port} is unavailable: {error}") from error
|
|
server.listen()
|
|
http_requests[port] = 0
|
|
|
|
def serve() -> None:
|
|
while True:
|
|
try:
|
|
connection, _address = server.accept()
|
|
except OSError:
|
|
return
|
|
with connection:
|
|
request = connection.recv(4096)
|
|
if request.startswith(b"GET /v1/models "):
|
|
http_requests[port] += 1
|
|
body = b'{"data":[{"id":"slice-one-model","name":"Slice One Model"}]}'
|
|
connection.sendall(
|
|
b"HTTP/1.1 200 OK\r\n"
|
|
+ f"Content-Length: {len(body)}\r\n".encode("ascii")
|
|
+ b"Content-Type: application/json\r\nConnection: close\r\n\r\n"
|
|
+ body
|
|
)
|
|
else:
|
|
connection.sendall(b"slice-1-ok")
|
|
|
|
threading.Thread(target=serve, daemon=True).start()
|
|
|
|
|
|
def _prepare_state(session: Any, context: str, lm_studio_extension: Path) -> Path:
|
|
agent = session.state / "pi-sbx" / context / "agent"
|
|
sessions = session.state / "pi-sbx" / context / "sessions"
|
|
cache = session.cache / "pi-sbx" / context
|
|
for path in (agent, sessions, cache):
|
|
path.mkdir(parents=True, mode=0o700)
|
|
os.chmod(path, 0o700)
|
|
settings = isolated_settings("lm-studio", lm_studio_extension=lm_studio_extension)
|
|
(agent / "settings.json").write_text(
|
|
json.dumps(settings, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(agent / "settings.json", 0o600)
|
|
instructions = session.home / ".pi" / "agent" / "AGENTS.md"
|
|
instructions.parent.mkdir(parents=True, mode=0o700)
|
|
instructions.write_text("fixture instructions\n", encoding="utf-8")
|
|
return agent
|
|
|
|
|
|
def _write_probe(
|
|
repository: Path,
|
|
other_tmp: Path,
|
|
other_state: Path,
|
|
other_repository: Path,
|
|
installed_extension: Path,
|
|
unselected_config: Path,
|
|
) -> Path:
|
|
probe = repository / "slice_one_probe.py"
|
|
probe.write_text(
|
|
"""\
|
|
import json, os, socket
|
|
from pathlib import Path
|
|
|
|
ports = {}
|
|
for port in (1234, 1235):
|
|
try:
|
|
with socket.create_connection((\"127.0.0.1\", port), timeout=1) as connection:
|
|
connection.sendall(b\"SLICE_ONE_PROBE\")
|
|
ports[str(port)] = connection.recv(32).decode(\"utf-8\")
|
|
except Exception as error:
|
|
ports[str(port)] = type(error).__name__
|
|
|
|
def can_read(path):
|
|
try:
|
|
Path(path).read_bytes()
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
def can_write(path):
|
|
try:
|
|
Path(path).write_text(\"changed\", encoding=\"utf-8\")
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
print(json.dumps({
|
|
\"ports\": ports,
|
|
\"rawCredentials\": sorted(name for name in %r if name in os.environ),
|
|
\"otherTmpReadable\": can_read(%r),
|
|
\"otherStateReadable\": can_read(%r),
|
|
\"otherRepositoryReadable\": can_read(%r),
|
|
\"installedExtensionWritable\": can_write(%r),
|
|
\"unselectedConfigReadable\": can_read(%r),
|
|
\"workdirWritable\": can_write(Path.cwd() / \"probe-write.txt\"),
|
|
}, sort_keys=True))
|
|
"""
|
|
% (
|
|
tuple(sorted(RAW_CANARIES)),
|
|
str(other_tmp / "private.txt"),
|
|
str(other_state),
|
|
str(other_repository),
|
|
str(installed_extension),
|
|
str(unselected_config),
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return probe
|
|
|
|
|
|
def _write_refresh_runner(repository: Path, installed_extension: Path) -> Path:
|
|
runner = repository / "lm_studio_refresh.mjs"
|
|
runner.write_text(
|
|
f"""\
|
|
import registerLmStudio from {json.dumps(installed_extension.as_uri())};
|
|
let refreshModels;
|
|
registerLmStudio({{
|
|
registerProvider(name, config) {{
|
|
if (name !== "lm-studio" || typeof config.refreshModels !== "function") {{
|
|
throw new Error("managed LM Studio provider registration is invalid");
|
|
}}
|
|
refreshModels = config.refreshModels;
|
|
}},
|
|
}});
|
|
const models = await refreshModels({{
|
|
allowNetwork: true,
|
|
signal: AbortSignal.timeout(3000),
|
|
}});
|
|
console.log(JSON.stringify({{ modelIds: models.map((model) => model.id) }}));
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
return runner
|
|
|
|
|
|
def _run(
|
|
nono: str,
|
|
profile: Path,
|
|
repository: Path,
|
|
environment: dict[str, str],
|
|
command: list[str],
|
|
) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
[
|
|
nono,
|
|
"run",
|
|
"--profile",
|
|
str(profile),
|
|
"--allow-cwd",
|
|
"--workdir",
|
|
str(repository),
|
|
"--",
|
|
*command,
|
|
],
|
|
cwd=repository,
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
input="",
|
|
check=False,
|
|
timeout=30,
|
|
)
|
|
|
|
|
|
def _last_json(stdout: str) -> dict[str, Any]:
|
|
for line in reversed(stdout.splitlines()):
|
|
try:
|
|
value = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(value, dict):
|
|
return value
|
|
raise HostGateError(f"command produced no JSON object: {stdout[-1000:]}")
|
|
|
|
|
|
def run_host_gate() -> dict[str, Any]:
|
|
if sys.platform != "linux":
|
|
raise HostGateError("Slice 1 host gate requires Linux")
|
|
nono = shutil.which("nono", path="/usr/bin:/bin")
|
|
if nono is None:
|
|
raise HostGateError("Nono is unavailable")
|
|
version = subprocess.run(
|
|
[nono, "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=10,
|
|
env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
|
|
)
|
|
if version.returncode != 0 or not version.stdout.strip().startswith("nono 0.74."):
|
|
raise HostGateError(f"Nono 0.74.x is required; observed {version.stdout.strip()!r}")
|
|
if not PI_CLI.is_file():
|
|
raise HostGateError(f"Pi CLI is unavailable: {PI_CLI}")
|
|
managed_lm_studio_source = _validate_lm_studio_sources()
|
|
|
|
outcomes: dict[str, Any] = {}
|
|
active_state_before = _active_state_snapshot()
|
|
listener_requests: dict[int, int] = {}
|
|
with ExitStack() as listeners:
|
|
_listener(1234, listeners, listener_requests)
|
|
_listener(1235, listeners, listener_requests)
|
|
# Nono's command-policy supervisor uses a Unix socket under TMPDIR;
|
|
# keep the isolated fixture path below Linux SUN_LEN.
|
|
with isolated_environment(prefix="s1-") as fixture:
|
|
installed_extension = fixture.root / "installed" / "lm-studio" / "index.ts"
|
|
installed_extension.parent.mkdir(parents=True, mode=0o700)
|
|
installed_extension.write_bytes(managed_lm_studio_source)
|
|
os.chmod(installed_extension, 0o400)
|
|
unselected_config = fixture.root / "unselected" / "auth.json"
|
|
unselected_config.parent.mkdir(parents=True, mode=0o700)
|
|
unselected_config.write_text("fixture credential config\n", encoding="utf-8")
|
|
os.chmod(unselected_config, 0o600)
|
|
sessions = {name: fixture.session(name) for name in ("personal", "work", "other")}
|
|
for session in sessions.values():
|
|
(session.tmp / "private.txt").write_text("private\n", encoding="utf-8")
|
|
(session.repo / "private.txt").write_text("private\n", encoding="utf-8")
|
|
personal_environment = sessions["personal"].environment(fixture.env)
|
|
_validate_static_profiles(nono, personal_environment)
|
|
_prove_provider_fragment_is_not_launchable(nono, personal_environment)
|
|
_prove_nono_composition(
|
|
nono,
|
|
sessions["personal"],
|
|
personal_environment,
|
|
)
|
|
other_state = sessions["other"].state / "pi-sbx" / "other"
|
|
other_state.mkdir(parents=True, mode=0o700)
|
|
(other_state / "private.txt").write_text("private\n", encoding="utf-8")
|
|
for context in ("personal", "work"):
|
|
session = sessions[context]
|
|
other = sessions["other"]
|
|
launch_temporary_directory = Path(
|
|
tempfile.mkdtemp(prefix=f"s1{context[0]}-", dir="/tmp")
|
|
)
|
|
os.chmod(launch_temporary_directory, 0o700)
|
|
listeners.callback(
|
|
shutil.rmtree,
|
|
launch_temporary_directory,
|
|
ignore_errors=True,
|
|
)
|
|
launch_environment = session.environment(fixture.env)
|
|
launch_environment["TMPDIR"] = str(launch_temporary_directory)
|
|
_prepare_state(session, context, installed_extension)
|
|
probe = _write_probe(
|
|
session.repo,
|
|
other.tmp,
|
|
other_state / "private.txt",
|
|
other.repo / "private.txt",
|
|
installed_extension,
|
|
unselected_config,
|
|
)
|
|
refresh_runner = _write_refresh_runner(session.repo, installed_extension)
|
|
profile = compose_profile(
|
|
load_profile(PROFILES / f"pi-core-{context}.json"),
|
|
load_profile(PROFILES / "provider-lm-studio.json"),
|
|
workdir_access="read" if context == "personal" else "readwrite",
|
|
temporary_directory=launch_temporary_directory,
|
|
lm_studio_extension=installed_extension,
|
|
)
|
|
profile_path = fixture.root / f"generated-{context}.json"
|
|
profile_path.write_text(
|
|
json.dumps(profile, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(profile_path, 0o600)
|
|
validation = subprocess.run(
|
|
[nono, "profile", "validate", "--strict", str(profile_path)],
|
|
env=fixture.env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
if validation.returncode != 0:
|
|
raise HostGateError(validation.stderr + validation.stdout)
|
|
|
|
probe_environment = dict(launch_environment)
|
|
probe_environment.update(RAW_CANARIES)
|
|
probe_result = _run(
|
|
nono,
|
|
profile_path,
|
|
session.repo,
|
|
probe_environment,
|
|
["/usr/bin/python3", str(probe)],
|
|
)
|
|
if probe_result.returncode != 0:
|
|
raise HostGateError(probe_result.stderr + probe_result.stdout)
|
|
observed = _last_json(probe_result.stdout)
|
|
expected_write = context == "work"
|
|
if observed.get("ports", {}).get("1234") != "slice-1-ok":
|
|
raise HostGateError(f"{context}: selected LM Studio port was denied")
|
|
if observed.get("ports", {}).get("1235") == "slice-1-ok":
|
|
raise HostGateError(f"{context}: adjacent loopback port was allowed")
|
|
if observed.get("rawCredentials"):
|
|
raise HostGateError(f"{context}: raw credential names reached the sandbox")
|
|
if (
|
|
observed.get("otherTmpReadable")
|
|
or observed.get("otherStateReadable")
|
|
or observed.get("otherRepositoryReadable")
|
|
):
|
|
raise HostGateError(f"{context}: cross-context data was readable")
|
|
if observed.get("installedExtensionWritable"):
|
|
raise HostGateError(f"{context}: installed extension was writable")
|
|
if observed.get("unselectedConfigReadable"):
|
|
raise HostGateError(f"{context}: unselected configuration was readable")
|
|
if observed.get("workdirWritable") is not expected_write:
|
|
raise HostGateError(f"{context}: workdir access did not match the declaration")
|
|
|
|
rpc = _run(
|
|
nono,
|
|
profile_path,
|
|
session.repo,
|
|
launch_environment,
|
|
[
|
|
"/usr/bin/node",
|
|
str(PI_CLI),
|
|
"--mode",
|
|
"rpc",
|
|
"--no-session",
|
|
"--no-skills",
|
|
],
|
|
)
|
|
# EOF exits RPC after the process has initialized. A successful
|
|
# startup may produce no response because no command was sent.
|
|
if rpc.returncode != 0:
|
|
raise HostGateError(f"{context}: Pi startup failed: {rpc.stderr}")
|
|
|
|
requests_before_refresh = listener_requests[1234]
|
|
model_refresh = _run(
|
|
nono,
|
|
profile_path,
|
|
session.repo,
|
|
launch_environment,
|
|
[
|
|
"/usr/bin/node",
|
|
"--experimental-strip-types",
|
|
str(refresh_runner),
|
|
],
|
|
)
|
|
if model_refresh.returncode != 0:
|
|
raise HostGateError(
|
|
f"{context}: LM Studio model refresh failed: "
|
|
f"{model_refresh.stderr}{model_refresh.stdout}"
|
|
)
|
|
refreshed_models = _last_json(model_refresh.stdout)
|
|
if refreshed_models.get("modelIds") != ["slice-one-model"]:
|
|
raise HostGateError(f"{context}: LM Studio model response was not mapped")
|
|
if listener_requests[1234] != requests_before_refresh + 1:
|
|
raise HostGateError(
|
|
f"{context}: LM Studio refresh did not request the exact model endpoint"
|
|
)
|
|
outcomes[context] = observed
|
|
|
|
if _active_state_snapshot() != active_state_before:
|
|
raise HostGateError("Slice 1 verification mutated active Pi or Nono state")
|
|
tested_profiles = sorted(PROFILES.glob("pi-core-*.json")) + sorted(
|
|
PROFILES.glob("provider-*.json")
|
|
)
|
|
return {
|
|
"suite": "slice-1-host",
|
|
"status": "passed",
|
|
"nonoVersion": version.stdout.strip().split()[-1],
|
|
"contexts": sorted(outcomes),
|
|
"loopbackChecks": 4,
|
|
"isolationChecks": 12,
|
|
"piStartupChecks": 2,
|
|
"providerRequestChecks": 2,
|
|
"directLaunchDenialChecks": 1,
|
|
"staticProfileChecks": len(tested_profiles),
|
|
"compositionChecks": 3,
|
|
"commandPolicyChecks": 2,
|
|
"profileDigests": {
|
|
profile.name: hashlib.sha256(profile.read_bytes()).hexdigest()
|
|
for profile in tested_profiles
|
|
},
|
|
"lmStudioDigest": hashlib.sha256(managed_lm_studio_source).hexdigest(),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
active_before: dict[str, tuple[str, str, int]] | None = None
|
|
try:
|
|
active_before = _active_state_snapshot()
|
|
result = run_host_gate()
|
|
if _active_state_snapshot() != active_before:
|
|
raise HostGateError("Slice 1 verification mutated active Pi or Nono state")
|
|
print(json.dumps(result, sort_keys=True))
|
|
except (HostGateError, OSError, subprocess.SubprocessError, ValueError) as error:
|
|
if active_before is not None:
|
|
try:
|
|
if _active_state_snapshot() != active_before:
|
|
error = HostGateError(
|
|
"Slice 1 verification mutated active Pi or Nono state after "
|
|
f"failing: {error}"
|
|
)
|
|
except OSError as snapshot_error:
|
|
error = HostGateError(
|
|
f"Could not verify active state after failure: {snapshot_error}; "
|
|
f"original failure: {error}"
|
|
)
|
|
print(json.dumps({"suite": "slice-1-host", "status": "failed", "error": str(error)}))
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|