252 lines
9.7 KiB
Python
252 lines
9.7 KiB
Python
"""Trusted Slice 1 policy composition contract used by capability tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
|
|
PROVIDER_PROFILE_NAMES = {
|
|
"anthropic": "provider-anthropic",
|
|
"lm-studio": "provider-lm-studio",
|
|
"morph": "provider-morph",
|
|
"openai": "provider-openai",
|
|
}
|
|
PROVIDER_CREDENTIAL_DEFINITIONS = {
|
|
"anthropic": {
|
|
"upstream": "https://api.anthropic.com",
|
|
"credential_key": "ANTHROPIC_API_KEY",
|
|
"inject_header": "x-api-key",
|
|
"credential_format": "{}",
|
|
"env_var": "ANTHROPIC_API_KEY",
|
|
},
|
|
"morph": {
|
|
"upstream": "https://api.morphllm.com",
|
|
"credential_key": "MORPH_API_KEY",
|
|
"inject_header": "Authorization",
|
|
"credential_format": "Bearer {}",
|
|
"env_var": "MORPH_API_KEY",
|
|
},
|
|
"openai": {
|
|
"upstream": "https://api.openai.com/v1",
|
|
"credential_key": "OPENAI_API_KEY",
|
|
"inject_header": "Authorization",
|
|
"credential_format": "Bearer {}",
|
|
"env_var": "OPENAI_API_KEY",
|
|
},
|
|
}
|
|
RAW_CREDENTIAL_VARIABLES = frozenset(
|
|
{
|
|
"ANTHROPIC_API_KEY",
|
|
"GH_TOKEN",
|
|
"GITHUB_TOKEN",
|
|
"MORPH_API_KEY",
|
|
"OPENAI_API_KEY",
|
|
"SSH_AUTH_SOCK",
|
|
}
|
|
)
|
|
_LM_REGISTRATION = re.compile(r"registerProvider\(\s*[\"']lm-studio[\"']")
|
|
_ROOT = Path(__file__).resolve().parents[4]
|
|
_PROFILES = _ROOT / "profiles" / "nono"
|
|
_PACKAGE_LOCK = _ROOT / "config" / "pi-package-lock.json"
|
|
|
|
|
|
class CompositionError(ValueError):
|
|
"""A reviewed core/provider pair cannot produce one narrow launch policy."""
|
|
|
|
|
|
def load_profile(path: Path) -> dict[str, Any]:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise CompositionError(f"profile is not an object: {path}")
|
|
return data
|
|
|
|
|
|
def compose_profile(
|
|
core: dict[str, Any],
|
|
provider: dict[str, Any],
|
|
*,
|
|
workdir_access: Literal["read", "readwrite"],
|
|
temporary_directory: Path,
|
|
lm_studio_extension: Path | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Materialize one reviewed core and one provider into a launch profile.
|
|
|
|
This intentionally does not model Nono inheritance. Nono preserves a blocked
|
|
network across inherited parents, so the trusted generator must replace the
|
|
core's fail-closed network block with exactly one reviewed provider route.
|
|
"""
|
|
|
|
profile = copy.deepcopy(core)
|
|
core_name = _profile_name(core)
|
|
provider_name = _profile_name(provider)
|
|
if core_name not in {"pi-core-personal", "pi-core-work"}:
|
|
raise CompositionError(f"unapproved core profile: {core_name}")
|
|
if core != load_profile(_PROFILES / f"{core_name}.json"):
|
|
raise CompositionError(f"core profile differs from reviewed source: {core_name}")
|
|
expected_provider = PROVIDER_PROFILE_NAMES.get(
|
|
provider_name.removeprefix("provider-")
|
|
)
|
|
if expected_provider != provider_name:
|
|
raise CompositionError(f"unapproved provider profile: {provider_name}")
|
|
if provider != load_profile(_PROFILES / f"{provider_name}.json"):
|
|
raise CompositionError(
|
|
f"provider profile differs from reviewed source: {provider_name}"
|
|
)
|
|
_validate_provider_network(provider_name, provider.get("network"))
|
|
|
|
profile["extends"] = "default"
|
|
profile["meta"] = {
|
|
"name": f"generated-{core_name.removeprefix('pi-core-')}-{provider_name.removeprefix('provider-')}",
|
|
"description": "Generated Slice 1 composition fixture",
|
|
}
|
|
profile["workdir"] = {"access": workdir_access}
|
|
profile["network"] = copy.deepcopy(provider["network"])
|
|
|
|
environment = profile.get("environment")
|
|
if not isinstance(environment, dict):
|
|
raise CompositionError("core profile has no environment policy")
|
|
denied = set(environment.get("deny_vars", []))
|
|
if denied != RAW_CREDENTIAL_VARIABLES:
|
|
raise CompositionError("core raw-credential deny list is incomplete")
|
|
|
|
filesystem = profile.get("filesystem")
|
|
if not isinstance(filesystem, dict):
|
|
raise CompositionError("core profile has no filesystem policy")
|
|
if not temporary_directory.is_absolute():
|
|
raise CompositionError("launch temporary directory must be absolute")
|
|
try:
|
|
metadata = temporary_directory.lstat()
|
|
resolved_temporary_directory = temporary_directory.resolve(strict=True)
|
|
except OSError as error:
|
|
raise CompositionError("launch temporary directory must already exist") from error
|
|
if (
|
|
temporary_directory.is_symlink()
|
|
or not stat.S_ISDIR(metadata.st_mode)
|
|
or resolved_temporary_directory != temporary_directory
|
|
or metadata.st_uid != os.getuid()
|
|
or stat.S_IMODE(metadata.st_mode) & 0o077
|
|
or resolved_temporary_directory in {Path("/"), Path("/tmp"), Path.home()}
|
|
):
|
|
raise CompositionError(
|
|
"launch temporary directory must be one private owned directory"
|
|
)
|
|
mutable_directories = list(filesystem.get("allow", []))
|
|
if "$TMPDIR" in mutable_directories or "/tmp" in mutable_directories:
|
|
raise CompositionError("core profile contains a broad temporary grant")
|
|
mutable_directories.append(str(resolved_temporary_directory))
|
|
filesystem["allow"] = sorted(set(mutable_directories))
|
|
set_variables = dict(environment.get("set_vars", {}))
|
|
set_variables["TMPDIR"] = str(resolved_temporary_directory)
|
|
environment["set_vars"] = set_variables
|
|
if provider_name == "provider-lm-studio":
|
|
if lm_studio_extension is None:
|
|
raise CompositionError("LM Studio requires one installed extension path")
|
|
_validate_lm_studio_extension(lm_studio_extension)
|
|
reads = list(filesystem.get("read", []))
|
|
reads.append(str(lm_studio_extension))
|
|
filesystem["read"] = sorted(set(reads))
|
|
elif lm_studio_extension is not None:
|
|
raise CompositionError("non-LM providers cannot load the LM Studio extension")
|
|
|
|
return profile
|
|
|
|
|
|
def isolated_settings(
|
|
provider: str,
|
|
*,
|
|
lm_studio_extension: Path | None = None,
|
|
) -> dict[str, Any]:
|
|
if provider not in PROVIDER_PROFILE_NAMES:
|
|
raise CompositionError(f"unknown provider: {provider}")
|
|
extensions: list[str] = []
|
|
if provider == "lm-studio":
|
|
if lm_studio_extension is None:
|
|
raise CompositionError("LM Studio settings require one extension path")
|
|
_validate_lm_studio_extension(lm_studio_extension)
|
|
extensions.append(str(lm_studio_extension))
|
|
elif lm_studio_extension is not None:
|
|
raise CompositionError("LM Studio extension is forbidden for another provider")
|
|
return {
|
|
"defaultProjectTrust": "never",
|
|
"defaultTools": ["read", "edit", "write"],
|
|
"enableInstallTelemetry": False,
|
|
"extensions": extensions,
|
|
"packages": [],
|
|
"quietStartup": True,
|
|
}
|
|
|
|
|
|
def assert_single_lm_studio_registration(sources: list[Path]) -> None:
|
|
registrations: list[tuple[Path, int]] = []
|
|
for source in sources:
|
|
contents = source.read_text(encoding="utf-8")
|
|
if "http://localhost:1234" in contents:
|
|
raise CompositionError(f"legacy localhost LM Studio URL remains: {source}")
|
|
registrations.extend(
|
|
(source, match.start()) for match in _LM_REGISTRATION.finditer(contents)
|
|
)
|
|
if len(registrations) != 1:
|
|
raise CompositionError(
|
|
"expected exactly one LM Studio provider registration; found "
|
|
+ str(len(registrations))
|
|
)
|
|
|
|
|
|
def _validate_lm_studio_extension(path: Path) -> None:
|
|
if not path.is_absolute() or path.is_symlink() or not path.is_file():
|
|
raise CompositionError("LM Studio extension must be one absolute regular file")
|
|
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 CompositionError("package lock must contain one LM Studio entry")
|
|
expected_digest = entries[0].get("reviewed", {}).get("sha256")
|
|
observed_digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
if expected_digest != observed_digest:
|
|
raise CompositionError("LM Studio extension differs from reviewed lock source")
|
|
|
|
|
|
def _profile_name(profile: dict[str, Any]) -> str:
|
|
meta = profile.get("meta")
|
|
if not isinstance(meta, dict) or not isinstance(meta.get("name"), str):
|
|
raise CompositionError("profile has no name")
|
|
return meta["name"]
|
|
|
|
|
|
def _validate_provider_network(name: str, network: Any) -> None:
|
|
if not isinstance(network, dict) or network.get("block") is not False:
|
|
raise CompositionError(f"provider route is not enabled: {name}")
|
|
credentials = network.get("credentials", [])
|
|
connect_ports = network.get("connect_port", [])
|
|
if name == "provider-lm-studio":
|
|
if (
|
|
set(network) != {"block", "connect_port"}
|
|
or credentials
|
|
or connect_ports != [1234]
|
|
):
|
|
raise CompositionError("LM Studio must grant only connect port 1234")
|
|
return
|
|
expected_credential = name.removeprefix("provider-")
|
|
if set(network) != {
|
|
"block",
|
|
"allow_domain",
|
|
"credentials",
|
|
"custom_credentials",
|
|
}:
|
|
raise CompositionError(f"provider network contains an extra route: {name}")
|
|
if network["allow_domain"] or credentials != [expected_credential] or connect_ports:
|
|
raise CompositionError(f"provider must activate only its own route: {name}")
|
|
definitions = network.get("custom_credentials")
|
|
if definitions != {
|
|
expected_credential: PROVIDER_CREDENTIAL_DEFINITIONS[expected_credential]
|
|
}:
|
|
raise CompositionError(f"provider credential definition mismatch: {name}")
|