925 lines
38 KiB
Python
925 lines
38 KiB
Python
"""Offline contracts for the isolated GCP leoclean no-send service image."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ops import gcp_leoclean_nosend_package as package
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
REVISION = package._git_commit(ROOT)
|
|
IMAGE_DIGEST = "2" * 64
|
|
CLAIM_CEILING = "Offline package proof only; live GCP identity, database access, restart, and parity remain unproven."
|
|
|
|
|
|
def write_json(path: Path, value: object, mode: int = 0o600) -> None:
|
|
path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
path.chmod(mode)
|
|
|
|
|
|
def git_source_fixture(tmp_path: Path, files: dict[str, str]) -> tuple[Path, str, dict[str, str]]:
|
|
root = tmp_path / "source-repository"
|
|
root.mkdir()
|
|
subprocess.run(["git", "init", "--quiet", str(root)], check=True)
|
|
for relative, content in files.items():
|
|
path = root / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
subprocess.run(["git", "-C", str(root), "add", "--", *sorted(files)], check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(root),
|
|
"-c",
|
|
"user.name=fwazb",
|
|
"-c",
|
|
"user.email=fawaz.butt95@gmail.com",
|
|
"commit",
|
|
"--quiet",
|
|
"-m",
|
|
"Create source fixture",
|
|
],
|
|
check=True,
|
|
)
|
|
revision = package._git_commit(root)
|
|
expected = {relative: package.sha256_file(root / relative) for relative in files}
|
|
return root, revision, expected
|
|
|
|
|
|
def load_entrypoint_module(tmp_path: Path):
|
|
image_root = tmp_path / "entrypoint-image"
|
|
image_root.mkdir()
|
|
entrypoint_path = image_root / "entrypoint.py"
|
|
shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint_path)
|
|
shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
|
|
scripts = image_root / "scripts"
|
|
scripts.mkdir()
|
|
for relative in package.IDENTITY_CONTRACT_SOURCES:
|
|
shutil.copy2(ROOT / relative, image_root / relative)
|
|
spec = importlib.util.spec_from_file_location(f"leoclean_entrypoint_{tmp_path.name}", entrypoint_path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]:
|
|
artifact = tmp_path / "artifact"
|
|
artifact.mkdir()
|
|
(artifact / "bootstrap.py").write_text("#!/usr/bin/env python3\n", encoding="utf-8")
|
|
(artifact / "bootstrap.py").chmod(0o755)
|
|
hermes = artifact / "hermes-agent"
|
|
hermes.mkdir()
|
|
(hermes / "uv.lock").write_text(
|
|
'version = 1\n\n[[package]]\nname = "pyyaml"\nversion = "6.0.3"\n',
|
|
encoding="utf-8",
|
|
)
|
|
uv_lock_sha256 = package.sha256_file(hermes / "uv.lock")
|
|
write_json(
|
|
artifact / "runtime-lock.json",
|
|
{
|
|
"base_images": {
|
|
"python": package.PYTHON_IMAGE.removeprefix("docker.io/library/"),
|
|
"uv": package.UV_IMAGE,
|
|
},
|
|
"hermes": {
|
|
"commit": package.HERMES_COMMIT,
|
|
"uv_lock_sha256": uv_lock_sha256,
|
|
"patched_source_tree_sha256": "4" * 64,
|
|
},
|
|
},
|
|
mode=0o644,
|
|
)
|
|
profile = artifact / "profile-template"
|
|
profile.mkdir()
|
|
(profile / "config.yaml").write_text(
|
|
"""model:
|
|
provider: openrouter
|
|
default: anthropic/claude-sonnet-4-6
|
|
max_tokens: 8192
|
|
smart_model_routing: false
|
|
agent:
|
|
max_turns: 90
|
|
memory:
|
|
enabled: false
|
|
memory_enabled: false
|
|
user_profile_enabled: false
|
|
nudge_interval: 0
|
|
skills:
|
|
creation_nudge_interval: 0
|
|
platform_toolsets:
|
|
api_server: [livingip-leoclean-nosend]
|
|
known_plugin_toolsets:
|
|
api_server: []
|
|
mcp_servers: {}
|
|
display:
|
|
tool_progress: false
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
(profile / "config.yaml").chmod(0o600)
|
|
skills = profile / "skills" / "teleo-kb-bridge"
|
|
skills.mkdir(parents=True)
|
|
(skills / "SKILL.md").write_text("# fixture skill\n", encoding="utf-8")
|
|
write_json(
|
|
artifact / "runtime-contract.json",
|
|
{
|
|
"schema": "livingip.leocleanNoSendRuntimeContract.v1",
|
|
"transport_authority": "absent",
|
|
"toolset": {"allowed_tools": ["skills_list", "skill_view", "terminal"]},
|
|
"database": {
|
|
"direct_canonical_writes": "denied",
|
|
"direct_stage_table_writes": "denied",
|
|
"proposal_staging": "function_only",
|
|
"proposal_function": package.PROPOSAL_FUNCTION,
|
|
},
|
|
},
|
|
mode=0o644,
|
|
)
|
|
content = package.tree_manifest(artifact)
|
|
manifest = {
|
|
"schema": package.ARTIFACT_MANIFEST_SCHEMA,
|
|
"teleo_git_head": REVISION,
|
|
"hermes_commit": package.HERMES_COMMIT,
|
|
"claim_ceiling": CLAIM_CEILING,
|
|
"content": content,
|
|
}
|
|
write_json(artifact / "artifact-manifest.json", manifest, mode=0o644)
|
|
receipt = tmp_path / "receipt.json"
|
|
write_json(
|
|
receipt,
|
|
{
|
|
"schema": package.ARTIFACT_RECEIPT_SCHEMA,
|
|
"status": "pass",
|
|
"verification_mode": "release",
|
|
"teleo_git_head": REVISION,
|
|
"artifact_sha256": content["sha256"],
|
|
"hermes_commit": package.HERMES_COMMIT,
|
|
"hermes_source_sha256": "4" * 64,
|
|
"config_sha256": package.sha256_file(profile / "config.yaml"),
|
|
"runtime_probe": runtime_probe_fixture(package.sha256_file(profile / "config.yaml")),
|
|
"claim_ceiling": CLAIM_CEILING,
|
|
},
|
|
mode=0o644,
|
|
)
|
|
return artifact, receipt
|
|
|
|
|
|
def runtime_surface_fixture() -> dict[str, object]:
|
|
return copy.deepcopy(package.NO_SEND_TOOL_SURFACE)
|
|
|
|
|
|
def runtime_probe_fixture(config_sha256: str) -> dict[str, object]:
|
|
surface = runtime_surface_fixture()
|
|
return {
|
|
"schema": "livingip.leocleanNoSendRuntimeProbe.v1",
|
|
"status": "pass",
|
|
"config_sha256": config_sha256,
|
|
"gateway_adapter_count": 0,
|
|
"connected_platforms": [],
|
|
"python_version": package.PYTHON_VERSION,
|
|
"model": package.DEFAULT_MODEL,
|
|
"provider_credential_names_present": ["OPENROUTER_API_KEY"],
|
|
"provider_credential_sentinel_redaction": "pass",
|
|
"effective_policy": {
|
|
"cheap_model_route": None,
|
|
"memory_enabled": False,
|
|
"memory_nudge_interval": 0,
|
|
"skill_creation_nudge_interval": 0,
|
|
"smart_model_routing": False,
|
|
"user_profile_enabled": False,
|
|
},
|
|
"tool_surface": surface,
|
|
"lifecycle": {
|
|
"started": True,
|
|
"stopped": True,
|
|
"post_start_registry": surface,
|
|
"post_discovery_registry": surface,
|
|
},
|
|
}
|
|
|
|
|
|
def test_git_source_closure_accepts_exact_revision_and_ignores_unrelated_dirty_files(tmp_path: Path) -> None:
|
|
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
|
|
(root / "unrelated.txt").write_text("not in the image closure\n", encoding="utf-8")
|
|
|
|
package._verify_git_source_closure(root, revision, expected)
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["unstaged", "staged", "deleted", "mode"])
|
|
def test_git_source_closure_rejects_dirty_required_source(tmp_path: Path, mutation: str) -> None:
|
|
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
|
|
source = root / "package" / "source.py"
|
|
if mutation == "deleted":
|
|
source.unlink()
|
|
elif mutation == "mode":
|
|
source.chmod(0o755)
|
|
else:
|
|
source.write_text("reviewed = False\n", encoding="utf-8")
|
|
if mutation == "staged":
|
|
subprocess.run(["git", "-C", str(root), "add", "--", "package/source.py"], check=True)
|
|
|
|
with pytest.raises(package.PackageError, match="closure is dirty"):
|
|
package._verify_git_source_closure(root, revision, expected)
|
|
|
|
|
|
def test_git_source_closure_rejects_untracked_required_source(tmp_path: Path) -> None:
|
|
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
|
|
untracked = root / "package" / "validator.py"
|
|
untracked.write_text("unreviewed = True\n", encoding="utf-8")
|
|
expected["package/validator.py"] = package.sha256_file(untracked)
|
|
|
|
with pytest.raises(package.PackageError):
|
|
package._verify_git_source_closure(root, revision, expected)
|
|
|
|
|
|
def test_git_source_closure_rejects_head_and_hash_mismatch(tmp_path: Path) -> None:
|
|
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
|
|
(root / "unrelated.txt").write_text("second commit\n", encoding="utf-8")
|
|
subprocess.run(["git", "-C", str(root), "add", "unrelated.txt"], check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(root),
|
|
"-c",
|
|
"user.name=fwazb",
|
|
"-c",
|
|
"user.email=fawaz.butt95@gmail.com",
|
|
"commit",
|
|
"--quiet",
|
|
"-m",
|
|
"Advance source fixture",
|
|
],
|
|
check=True,
|
|
)
|
|
|
|
with pytest.raises(package.PackageError, match="reviewed revision"):
|
|
package._verify_git_source_closure(root, revision, expected)
|
|
|
|
current = package._git_commit(root)
|
|
wrong_hash = {"package/source.py": "9" * 64}
|
|
with pytest.raises(package.PackageError, match="package binding"):
|
|
package._verify_git_source_closure(root, current, wrong_hash)
|
|
|
|
|
|
def identity_fixture(tmp_path: Path, artifact: Path, receipt: Path) -> tuple[Path, dict[str, object]]:
|
|
runtime = package.validate_artifact(artifact, receipt, REVISION)
|
|
contract = package.build_expected_identity_contract(
|
|
artifact=artifact,
|
|
receipt_path=receipt,
|
|
runtime=runtime,
|
|
source_revision=REVISION,
|
|
source_root=ROOT,
|
|
)
|
|
fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1"
|
|
manifest = package.strict_identity_manifest.build_identity_manifest(
|
|
identity_runtime_contract=contract,
|
|
database_fingerprint_path=fixture / "leo-database-fingerprint-v1.json",
|
|
constitution_path=fixture / "leo-constitution-v1.json",
|
|
database_identity_path=fixture / "leo-database-identity-v1.json",
|
|
source_root=ROOT,
|
|
)
|
|
identity = tmp_path / "identity"
|
|
package.compile_identity_bundle(
|
|
manifest=manifest,
|
|
expected_identity_contract=contract,
|
|
source_root=ROOT,
|
|
output=identity,
|
|
)
|
|
return identity, contract
|
|
|
|
|
|
def fully_rehash_identity_bundle(identity: Path) -> None:
|
|
manifest_path = identity / "identity-manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
inputs_hash = package.canonical_sha256(manifest["identity_inputs"])
|
|
manifest["identity_inputs_sha256"] = inputs_hash
|
|
fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1"
|
|
fingerprint = package.strict_identity_manifest.load_json(
|
|
fixture / "leo-database-fingerprint-v1.json", "database fingerprint"
|
|
)
|
|
rules = package.strict_identity_manifest.canonical_rules(
|
|
package.strict_identity_manifest.load_json(fixture / "leo-constitution-v1.json", "constitution")
|
|
)
|
|
database_identity = package.strict_identity_manifest.load_json(
|
|
fixture / "leo-database-identity-v1.json", "database identity"
|
|
)
|
|
records = package.strict_identity_manifest.canonical_database_records(
|
|
database_identity,
|
|
fingerprint_sha256=fingerprint["fingerprint_sha256"],
|
|
)
|
|
provenance = package.strict_identity_manifest.database_identity_provenance(
|
|
database_identity,
|
|
fingerprint=fingerprint,
|
|
)
|
|
rendered = package.strict_identity_manifest.render_identity_views(
|
|
identity_inputs_sha256=inputs_hash,
|
|
database_fingerprint_sha256=fingerprint["fingerprint_sha256"],
|
|
database_provenance=provenance,
|
|
rules=rules,
|
|
records=records,
|
|
)
|
|
for name, content in rendered.items():
|
|
(identity / name).write_text(content, encoding="utf-8")
|
|
(identity / name).chmod(0o600)
|
|
manifest["compiled_views"][name]["sha256"] = package.sha256_file(identity / name)
|
|
manifest["compiled_views"][name]["generated_from_identity_inputs_sha256"] = inputs_hash
|
|
stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
|
|
manifest["manifest_sha256"] = package.canonical_sha256(stable)
|
|
write_json(manifest_path, manifest)
|
|
lock_path = identity / "identity-lock.json"
|
|
lock = json.loads(lock_path.read_text(encoding="utf-8"))
|
|
lock["manifest_sha256"] = manifest["manifest_sha256"]
|
|
lock["identity_inputs_sha256"] = inputs_hash
|
|
lock["compiled_views"] = {
|
|
name: {"sha256": package.sha256_file(identity / name)} for name in sorted(rendered)
|
|
}
|
|
write_json(lock_path, lock)
|
|
|
|
|
|
def fully_rehash_identity_contract(contract: dict[str, object]) -> None:
|
|
runtime = contract["runtime"]
|
|
assert isinstance(runtime, dict)
|
|
runtime["identity_runtime_sha256"] = package.canonical_sha256(package._identity_runtime_payload(contract))
|
|
stable = {key: value for key, value in contract.items() if key != "contract_sha256"}
|
|
contract["contract_sha256"] = package.canonical_sha256(stable)
|
|
|
|
|
|
def refresh_image_identity_binding(image_input: dict[str, object], identity: Path) -> None:
|
|
binding = image_input["identity"]
|
|
assert isinstance(binding, dict)
|
|
manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8"))
|
|
binding.update(
|
|
{
|
|
"bundle_sha256": package.identity_bundle_manifest(identity)["sha256"],
|
|
"manifest_sha256": manifest["manifest_sha256"],
|
|
"identity_inputs_sha256": manifest["identity_inputs_sha256"],
|
|
"identity_lock_sha256": package.sha256_file(identity / "identity-lock.json"),
|
|
"soul_sha256": package.sha256_file(identity / "SOUL.md"),
|
|
"identity_view_sha256": package.sha256_file(identity / "identity.json"),
|
|
}
|
|
)
|
|
stable = {key: value for key, value in image_input.items() if key != "input_sha256"}
|
|
image_input["input_sha256"] = package.canonical_sha256(stable)
|
|
|
|
|
|
def image_input_fixture(tmp_path: Path) -> dict[str, object]:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, _contract = identity_fixture(tmp_path, artifact, receipt)
|
|
return package.build_image_input(
|
|
artifact=artifact,
|
|
receipt=receipt,
|
|
identity=identity,
|
|
ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem",
|
|
source_revision=REVISION,
|
|
dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile",
|
|
entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py",
|
|
package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py",
|
|
contract_source_root=ROOT,
|
|
identity_source_root=ROOT,
|
|
)
|
|
|
|
|
|
def test_image_input_binds_exact_runtime_identity_and_staging_target(tmp_path: Path) -> None:
|
|
value = image_input_fixture(tmp_path)
|
|
package.validate_image_input(value)
|
|
|
|
assert value["target"] == {
|
|
"project": "teleo-501523",
|
|
"zone": "europe-west6-a",
|
|
"instance": "teleo-staging-1",
|
|
"service": "leoclean-gcp-nosend.service",
|
|
"service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",
|
|
"platform": "linux/amd64",
|
|
}
|
|
assert value["runtime"]["python_version"] == "3.11.9"
|
|
assert value["runtime"]["postgres_version"] == "16.14"
|
|
assert value["identity"]["canonical_authority_granted"] is False
|
|
database_policy = value["identity"]["runtime_contract"]["permissions"]["runtime_policy"]["database"]
|
|
assert database_policy == {
|
|
"direct_canonical_writes": "denied",
|
|
"direct_stage_table_writes": "denied",
|
|
"proposal_staging": "function_only",
|
|
"proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)",
|
|
"live_effective_permissions_proven": False,
|
|
}
|
|
assert value["base_images"]["postgres"].endswith(
|
|
"@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55"
|
|
)
|
|
|
|
|
|
def test_artifact_requires_release_receipt_exact_revision_and_unchanged_content(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
package.validate_artifact(artifact, receipt, REVISION)
|
|
|
|
development = json.loads(receipt.read_text(encoding="utf-8"))
|
|
development["verification_mode"] = "development"
|
|
write_json(receipt, development, mode=0o644)
|
|
with pytest.raises(package.PackageError, match="not release"):
|
|
package.validate_artifact(artifact, receipt, REVISION)
|
|
|
|
development["verification_mode"] = "release"
|
|
write_json(receipt, development, mode=0o644)
|
|
(artifact / "bootstrap.py").write_text("mutated\n", encoding="utf-8")
|
|
(artifact / "bootstrap.py").chmod(0o755)
|
|
with pytest.raises(package.PackageError, match="content manifest drifted"):
|
|
package.validate_artifact(artifact, receipt, REVISION)
|
|
|
|
|
|
def test_artifact_binds_runtime_probe_to_the_exact_packaged_config(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
value = json.loads(receipt.read_text(encoding="utf-8"))
|
|
value["runtime_probe"]["config_sha256"] = "9" * 64
|
|
write_json(receipt, value, mode=0o644)
|
|
|
|
with pytest.raises(package.PackageError, match="runtime probe config differs"):
|
|
package.validate_artifact(artifact, receipt, REVISION)
|
|
|
|
|
|
def test_identity_bundle_rejects_drift_extra_files_and_false_authority(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, contract = identity_fixture(tmp_path, artifact, receipt)
|
|
validate = lambda root: package.validate_identity_bundle( # noqa: E731
|
|
root,
|
|
expected_revision=REVISION,
|
|
expected_identity_contract=contract,
|
|
source_root=ROOT,
|
|
)
|
|
validate(identity)
|
|
|
|
(identity / "SOUL.md").write_text("drift\n", encoding="utf-8")
|
|
(identity / "SOUL.md").chmod(0o600)
|
|
with pytest.raises(package.PackageError, match="view derivation drifted"):
|
|
validate(identity)
|
|
|
|
identity, _contract = identity_fixture(tmp_path / "second", artifact, receipt)
|
|
(identity / "extra.txt").write_text("extra\n", encoding="utf-8")
|
|
with pytest.raises(package.PackageError, match="entry set"):
|
|
validate(identity)
|
|
|
|
identity, _contract = identity_fixture(tmp_path / "third", artifact, receipt)
|
|
manifest_path = identity / "identity-manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["identity_inputs"]["canonical_database"] = {
|
|
"authority": "synthetic_noncanonical_fixture",
|
|
"canonical_authority_granted": True,
|
|
}
|
|
stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
|
|
manifest["manifest_sha256"] = package.canonical_sha256(stable)
|
|
write_json(manifest_path, manifest)
|
|
lock_path = identity / "identity-lock.json"
|
|
lock = json.loads(lock_path.read_text(encoding="utf-8"))
|
|
lock["manifest_sha256"] = manifest["manifest_sha256"]
|
|
write_json(lock_path, lock)
|
|
with pytest.raises(package.PackageError, match="semantic/source validation"):
|
|
validate(identity)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "value"),
|
|
[
|
|
(("source_commit",), "2" * 40),
|
|
(("runtime", "hermes_source_sha256"), "7" * 64),
|
|
(("model", "default_model"), "attacker/model"),
|
|
(("skills", "content_sha256"), "8" * 64),
|
|
(("permissions", "runtime_policy", "general_network_egress", "denial_tested"), True),
|
|
],
|
|
)
|
|
def test_fully_rehashed_identity_runtime_mutations_fail_closed(
|
|
tmp_path: Path, path: tuple[str, ...], value: object
|
|
) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, contract = identity_fixture(tmp_path, artifact, receipt)
|
|
manifest_path = identity / "identity-manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
target = manifest["identity_inputs"]
|
|
for key in path[:-1]:
|
|
target = target[key]
|
|
target[path[-1]] = value
|
|
if path == ("source_commit",):
|
|
manifest["identity_inputs"]["runtime"]["teleo_git_head"] = value
|
|
write_json(manifest_path, manifest)
|
|
fully_rehash_identity_bundle(identity)
|
|
|
|
with pytest.raises(package.PackageError):
|
|
package.validate_identity_bundle(
|
|
identity,
|
|
expected_revision=REVISION,
|
|
expected_identity_contract=contract,
|
|
source_root=ROOT,
|
|
)
|
|
|
|
|
|
def test_custom_no_send_policy_requires_explicit_validator_authority(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, contract = identity_fixture(tmp_path, artifact, receipt)
|
|
manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8"))
|
|
|
|
with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="exact local readback policy"):
|
|
package.strict_identity_manifest.validate_identity_manifest(manifest)
|
|
package.strict_identity_manifest.validate_identity_manifest(
|
|
manifest,
|
|
expected_runtime_policy_value=contract["permissions"]["runtime_policy"],
|
|
)
|
|
with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="expected identity runtime policy"):
|
|
package.strict_identity_manifest.validate_identity_manifest(manifest, expected_runtime_policy_value={})
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["legacy_policy", "malformed_behavior_inputs"])
|
|
def test_runtime_contract_adapter_rejects_weaker_or_malformed_contracts(tmp_path: Path, mutation: str) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
_identity, contract = identity_fixture(tmp_path, artifact, receipt)
|
|
mutated = copy.deepcopy(contract)
|
|
if mutation == "legacy_policy":
|
|
mutated["permissions"]["runtime_policy"] = package.strict_identity_manifest.expected_runtime_policy()
|
|
else:
|
|
del mutated["behavior_inputs"]["tool_surface"]
|
|
fully_rehash_identity_contract(mutated)
|
|
|
|
with pytest.raises(package.strict_identity_manifest.IdentityManifestError):
|
|
package.strict_identity_manifest.identity_runtime_inputs_from_contract(mutated)
|
|
|
|
|
|
def test_identity_source_content_drift_fails_closed(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, contract = identity_fixture(tmp_path, artifact, receipt)
|
|
source_root = tmp_path / "source-root"
|
|
fixture_relative = Path("fixtures/working-leo/leo-identity-v1")
|
|
shutil.copytree(ROOT / fixture_relative, source_root / fixture_relative)
|
|
package.validate_identity_bundle(
|
|
identity,
|
|
expected_revision=REVISION,
|
|
expected_identity_contract=contract,
|
|
source_root=source_root,
|
|
)
|
|
constitution = source_root / fixture_relative / "leo-constitution-v1.json"
|
|
constitution.write_text(constitution.read_text(encoding="utf-8") + "\n", encoding="utf-8")
|
|
|
|
with pytest.raises(package.PackageError, match="semantic/source validation"):
|
|
package.validate_identity_bundle(
|
|
identity,
|
|
expected_revision=REVISION,
|
|
expected_identity_contract=contract,
|
|
source_root=source_root,
|
|
)
|
|
|
|
|
|
def test_identity_source_intermediate_symlink_fails_closed(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, contract = identity_fixture(tmp_path, artifact, receipt)
|
|
outside = tmp_path / "outside"
|
|
shutil.copytree(ROOT / "fixtures", outside / "fixtures")
|
|
source_root = tmp_path / "source-root"
|
|
source_root.mkdir()
|
|
(source_root / "fixtures").symlink_to(outside / "fixtures", target_is_directory=True)
|
|
|
|
with pytest.raises(package.PackageError, match="semantic/source validation"):
|
|
package.validate_identity_bundle(
|
|
identity,
|
|
expected_revision=REVISION,
|
|
expected_identity_contract=contract,
|
|
source_root=source_root,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["skills", "behavior_artifact_and_uv"])
|
|
def test_installed_validation_rederives_fully_rehashed_identity_contract(tmp_path: Path, mutation: str) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, _contract = identity_fixture(tmp_path, artifact, receipt)
|
|
image_input = package.build_image_input(
|
|
artifact=artifact,
|
|
receipt=receipt,
|
|
identity=identity,
|
|
ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem",
|
|
source_revision=REVISION,
|
|
dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile",
|
|
entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py",
|
|
package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py",
|
|
contract_source_root=ROOT,
|
|
identity_source_root=ROOT,
|
|
)
|
|
embedded = image_input["identity"]["runtime_contract"]
|
|
manifest_path = identity / "identity-manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
if mutation == "skills":
|
|
embedded["skills"] = {"content_sha256": "8" * 64, "file_count": 999}
|
|
manifest["identity_inputs"]["skills"] = copy.deepcopy(embedded["skills"])
|
|
else:
|
|
embedded["behavior_inputs"]["artifact_sha256"] = "8" * 64
|
|
embedded["behavior_inputs"]["uv_lock_sha256"] = "9" * 64
|
|
image_input["runtime"]["artifact_sha256"] = "8" * 64
|
|
image_input["runtime"]["uv_lock_sha256"] = "9" * 64
|
|
fully_rehash_identity_contract(embedded)
|
|
manifest["identity_inputs"]["runtime"]["identity_runtime_sha256"] = embedded["runtime"][
|
|
"identity_runtime_sha256"
|
|
]
|
|
write_json(manifest_path, manifest)
|
|
fully_rehash_identity_bundle(identity)
|
|
refresh_image_identity_binding(image_input, identity)
|
|
|
|
package.validate_image_input(image_input)
|
|
with pytest.raises(package.PackageError):
|
|
package.validate_installed_runtime_identity(
|
|
image_input,
|
|
artifact=artifact,
|
|
receipt=receipt,
|
|
identity=identity,
|
|
contract_source_root=ROOT,
|
|
identity_source_root=ROOT,
|
|
)
|
|
|
|
|
|
def test_release_requires_exact_digest_repository_and_strict_schema(tmp_path: Path) -> None:
|
|
image_input = image_input_fixture(tmp_path)
|
|
reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
|
|
release = package.build_release_descriptor(image_input, reference)
|
|
package.validate_release_descriptor(release)
|
|
|
|
assert release["image"]["reference"] == reference
|
|
assert release["image"]["digest"] == f"sha256:{IMAGE_DIGEST}"
|
|
assert release["image"]["platform"] == "linux/amd64"
|
|
|
|
with pytest.raises(package.PackageError, match="sha256 digest"):
|
|
package.build_release_descriptor(image_input, f"{package.IMAGE_REPOSITORY}:latest")
|
|
with pytest.raises(package.PackageError, match="staging repository"):
|
|
package.build_release_descriptor(
|
|
image_input,
|
|
f"europe-west6-docker.pkg.dev/teleo-501523/teleo/other@sha256:{IMAGE_DIGEST}",
|
|
)
|
|
|
|
extra = copy.deepcopy(release)
|
|
extra["unexpected"] = True
|
|
with pytest.raises(package.PackageError, match="fields are not exact"):
|
|
package.validate_release_descriptor(extra)
|
|
|
|
wrong_label = copy.deepcopy(release)
|
|
wrong_label["image"]["labels"]["livingip.artifact.sha256"] = "9" * 64
|
|
stable = {key: value for key, value in wrong_label.items() if key != "release_sha256"}
|
|
wrong_label["release_sha256"] = package.canonical_sha256(stable)
|
|
with pytest.raises(package.PackageError, match="labels are not exact"):
|
|
package.validate_release_descriptor(wrong_label)
|
|
|
|
|
|
def test_rendered_unit_is_isolated_digest_pinned_and_has_no_credential_or_transport_surface(tmp_path: Path) -> None:
|
|
image_input = image_input_fixture(tmp_path)
|
|
release = package.build_release_descriptor(
|
|
image_input,
|
|
f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}",
|
|
)
|
|
unit = package.render_systemd_unit(release)
|
|
|
|
assert "leoclean-gcp-nosend" in unit
|
|
assert f"@sha256:{IMAGE_DIGEST}" in unit
|
|
assert "--pull=never" in unit
|
|
assert "--read-only" in unit
|
|
assert "--cap-drop=ALL" in unit
|
|
assert "--cap-add=CHOWN" in unit
|
|
assert "--cap-add=SETGID" in unit
|
|
assert "--cap-add=SETUID" in unit
|
|
assert "--cap-add=SETPCAP" in unit
|
|
assert "--network=bridge" in unit
|
|
assert "--tmpfs=/run/leoclean-profile:" in unit
|
|
assert "--tmpfs=/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" in unit
|
|
assert "NoNewPrivileges=yes" in unit
|
|
assert "EnvironmentFile" not in unit
|
|
assert "LoadCredential" not in unit
|
|
assert "docker.sock" not in unit
|
|
assert "--publish" not in unit
|
|
assert "teleo-prod-1" not in unit
|
|
assert "77.42.65.182" not in unit
|
|
assert "telegram" not in unit.casefold()
|
|
|
|
|
|
def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_overwrite(tmp_path: Path) -> None:
|
|
artifact, receipt = artifact_fixture(tmp_path)
|
|
identity, _contract = identity_fixture(tmp_path, artifact, receipt)
|
|
output = tmp_path / "context"
|
|
result = package.prepare_context(
|
|
repo_root=ROOT,
|
|
artifact=artifact,
|
|
receipt=receipt,
|
|
identity=identity,
|
|
identity_source_root=ROOT,
|
|
output=output,
|
|
source_revision=REVISION,
|
|
)
|
|
|
|
package.validate_image_input(result)
|
|
assert sorted(path.name for path in output.iterdir()) == [
|
|
"Dockerfile",
|
|
"artifact",
|
|
"artifact-receipt.json",
|
|
"build-args.json",
|
|
"cloudsql-server-ca.pem",
|
|
"entrypoint.py",
|
|
"identity",
|
|
"identity-sources",
|
|
"image-input.json",
|
|
"package_contract.py",
|
|
"scripts",
|
|
]
|
|
assert all((output / "identity" / name).stat().st_mode & 0o777 == 0o600 for name in package.IDENTITY_FILES)
|
|
encoded = (output / "image-input.json").read_text(encoding="utf-8").casefold()
|
|
assert "telegram_bot_token" not in encoded
|
|
assert "postgres-password" not in encoded
|
|
|
|
with pytest.raises(package.PackageError, match="already exists"):
|
|
package.prepare_context(
|
|
repo_root=ROOT,
|
|
artifact=artifact,
|
|
receipt=receipt,
|
|
identity=identity,
|
|
identity_source_root=ROOT,
|
|
output=output,
|
|
source_revision=REVISION,
|
|
)
|
|
|
|
|
|
def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install() -> None:
|
|
dockerfile = (ROOT / "docker" / "leoclean-nosend" / "Dockerfile").read_text(encoding="utf-8")
|
|
from_lines = [line for line in dockerfile.splitlines() if line.startswith("FROM ")]
|
|
|
|
assert len(from_lines) == 3
|
|
assert all("@sha256:" in line for line in from_lines)
|
|
assert "# syntax=" not in dockerfile
|
|
assert "postgres:16.14-bookworm@sha256:92620d" in dockerfile
|
|
assert "python:3.11.9-slim-bookworm@sha256:8fb099" in dockerfile
|
|
assert "uv:0.9.30@sha256:538e0b" in dockerfile
|
|
assert "apt-get" not in dockerfile
|
|
assert "pip install" not in dockerfile
|
|
assert "uv sync" in dockerfile
|
|
assert "--frozen" in dockerfile
|
|
assert "--no-editable" in dockerfile
|
|
assert "HEALTHCHECK" in dockerfile
|
|
assert "verify-build" in dockerfile
|
|
assert "identity -type f -exec chmod 0444" in dockerfile
|
|
|
|
|
|
def test_pid_one_command_line_accepts_only_native_and_exact_x86_64_binfmt_wrappers() -> None:
|
|
native = [b"/runtime/python", b"/runtime/bootstrap.py", b"service", b""]
|
|
binfmt = [native[0], *native]
|
|
|
|
assert package.pid_one_command_line_is_exact(native, native)
|
|
assert package.pid_one_command_line_is_exact(binfmt, native)
|
|
assert package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *binfmt], native)
|
|
assert package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64-static", *binfmt], native)
|
|
|
|
assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-aarch64", *binfmt], native)
|
|
assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *native], native)
|
|
assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *binfmt, b"--extra"], native)
|
|
assert not package.pid_one_command_line_is_exact(native, native[:-1])
|
|
|
|
|
|
def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pid_one() -> None:
|
|
source = (ROOT / "docker" / "leoclean-nosend" / "entrypoint.py").read_text(encoding="utf-8")
|
|
|
|
assert "os.setgroups([])" in source
|
|
assert "os.setgid(contract.RUNTIME_GID)" in source
|
|
assert "os.setuid(contract.RUNTIME_UID)" in source
|
|
assert "pr_capbset_drop = 24" in source
|
|
assert "reverse=True" in source
|
|
assert source.index("os.chown(runtime_identity, 0, 0)") < source.index(
|
|
"os.chown(PROFILE, contract.RUNTIME_UID, contract.RUNTIME_GID)"
|
|
)
|
|
assert 'value.get("process_id") == 1' in source
|
|
assert "contract.pid_one_command_line_is_exact(cmdline, expected_native)" in source
|
|
assert "contract.NO_SEND_TOOL_SURFACE" in source
|
|
assert 'image_input["runtime"]["config_sha256"]' in source
|
|
assert "gateway_adapter_count" in source
|
|
assert "os.execve" in source
|
|
assert "OPENROUTER_API_KEY" not in source
|
|
assert "TELEGRAM_BOT_TOKEN" not in source
|
|
|
|
|
|
def test_entrypoint_readiness_binds_packaged_and_live_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
entrypoint = load_entrypoint_module(tmp_path)
|
|
profile = tmp_path / "profile"
|
|
profile.mkdir()
|
|
config = profile / "config.yaml"
|
|
config.write_text("model: reviewed\n", encoding="utf-8")
|
|
expected = entrypoint.contract.sha256_file(config)
|
|
image_input = {"runtime": {"config_sha256": expected}}
|
|
readiness = {
|
|
"config_sha256": expected,
|
|
"tool_surface": copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE),
|
|
}
|
|
monkeypatch.setattr(entrypoint, "PROFILE", profile)
|
|
|
|
entrypoint._validate_readiness_behavior(image_input, readiness)
|
|
|
|
wrong_receipt = copy.deepcopy(readiness)
|
|
wrong_receipt["config_sha256"] = "9" * 64
|
|
with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
|
|
entrypoint._validate_readiness_behavior(image_input, wrong_receipt)
|
|
|
|
config.write_text("model: drifted\n", encoding="utf-8")
|
|
with pytest.raises(entrypoint.EntrypointError, match="profile config drifted"):
|
|
entrypoint._validate_readiness_behavior(image_input, readiness)
|
|
|
|
drift_hash = entrypoint.contract.sha256_file(config)
|
|
both_drifted = {**readiness, "config_sha256": drift_hash}
|
|
with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
|
|
entrypoint._validate_readiness_behavior(image_input, both_drifted)
|
|
|
|
config.unlink()
|
|
with pytest.raises(entrypoint.EntrypointError, match="missing or unsafe"):
|
|
entrypoint._validate_readiness_behavior(image_input, readiness)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "value"),
|
|
[
|
|
("extra", True),
|
|
("registry_sealed", False),
|
|
("send_message_present", True),
|
|
("terminal_handler", "unrestricted"),
|
|
("terminal_restricted_to", "shell"),
|
|
("toolset", "default"),
|
|
("tools", ["skill_view", "skills_list", "terminal", "send_message"]),
|
|
("plugins", []),
|
|
],
|
|
)
|
|
def test_entrypoint_readiness_rejects_exact_tool_surface_drift(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
mutation: str,
|
|
value: object,
|
|
) -> None:
|
|
entrypoint = load_entrypoint_module(tmp_path)
|
|
profile = tmp_path / "profile"
|
|
profile.mkdir()
|
|
config = profile / "config.yaml"
|
|
config.write_text("model: reviewed\n", encoding="utf-8")
|
|
expected = entrypoint.contract.sha256_file(config)
|
|
image_input = {"runtime": {"config_sha256": expected}}
|
|
surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
|
|
surface[mutation] = value
|
|
readiness = {"config_sha256": expected, "tool_surface": surface}
|
|
monkeypatch.setattr(entrypoint, "PROFILE", profile)
|
|
|
|
with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
|
|
entrypoint._validate_readiness_behavior(image_input, readiness)
|
|
|
|
|
|
def test_entrypoint_readiness_rejects_plugin_field_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
entrypoint = load_entrypoint_module(tmp_path)
|
|
profile = tmp_path / "profile"
|
|
profile.mkdir()
|
|
config = profile / "config.yaml"
|
|
config.write_text("model: reviewed\n", encoding="utf-8")
|
|
expected = entrypoint.contract.sha256_file(config)
|
|
monkeypatch.setattr(entrypoint, "PROFILE", profile)
|
|
|
|
for field, value in (
|
|
("source", "project"),
|
|
("tools", ["send_message"]),
|
|
("hooks", ["pre_llm_call"]),
|
|
("enabled", False),
|
|
("error", "load failed"),
|
|
):
|
|
surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
|
|
surface["plugins"][0][field] = value
|
|
readiness = {"config_sha256": expected, "tool_surface": surface}
|
|
with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
|
|
entrypoint._validate_readiness_behavior(
|
|
{"runtime": {"config_sha256": expected}},
|
|
readiness,
|
|
)
|
|
|
|
|
|
def test_entrypoint_loads_adjacent_contract_with_python_safe_path(tmp_path: Path) -> None:
|
|
image_root = tmp_path / "image"
|
|
image_root.mkdir()
|
|
entrypoint = image_root / "entrypoint.py"
|
|
shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint)
|
|
shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
|
|
scripts = image_root / "scripts"
|
|
scripts.mkdir()
|
|
for relative in package.IDENTITY_CONTRACT_SOURCES:
|
|
shutil.copy2(ROOT / relative, image_root / relative)
|
|
environment = {**os.environ, "PYTHONSAFEPATH": "1"}
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-P", str(entrypoint), "verify"],
|
|
cwd=tmp_path,
|
|
env=environment,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
|
|
assert result.returncode == 65
|
|
assert json.loads(result.stderr) == {"error": "container_contract_failed", "status": "fail"}
|
|
assert "Traceback" not in result.stderr
|