186 lines
8.1 KiB
Python
186 lines
8.1 KiB
Python
"""Focused unit coverage for the privileged runtime publication seam."""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tht import runtime_config_lease_io as lease_io
|
|
from tht.config import ConfigError, _read_runtime_config_source
|
|
|
|
|
|
def _manifest() -> dict:
|
|
return {
|
|
"version": 1, "workspace_id": "abc", "workspace_revision": "a" * 40,
|
|
"descriptor_git_blob": "b" * 40, "descriptor_sha256": "c" * 64,
|
|
"descriptor_dev": "1", "descriptor_ino": "2", "config_sha256": "d" * 64,
|
|
"config_dwh_binding": {
|
|
"workspace_id": "abc", "config_fingerprint": "e", "input_fingerprint": "f"
|
|
},
|
|
"config_dev": "1", "config_ino": "3", "config_size": "4",
|
|
"config_mode": "400", "config_uid": str(os.getuid()), "config_nlink": "1",
|
|
"directory_identities": [{"path": "/", "dev": "1", "ino": "1", "mode": "755", "uid": "0"}],
|
|
}
|
|
|
|
|
|
def test_strict_manifest_rejects_unknown_or_missing_fields():
|
|
value = _manifest()
|
|
assert lease_io.strict_manifest(value) is value
|
|
with pytest.raises(RuntimeError):
|
|
lease_io.strict_manifest({**value, "unexpected": True})
|
|
missing = dict(value)
|
|
del missing["config_sha256"]
|
|
with pytest.raises(RuntimeError):
|
|
lease_io.strict_manifest(missing)
|
|
|
|
|
|
def test_private_alias_is_darwin_only(monkeypatch):
|
|
monkeypatch.setattr(lease_io.sys, "platform", "linux")
|
|
assert lease_io._canonical_root("/tmp/runtime") == "/tmp/runtime"
|
|
assert lease_io._canonical_root("/var/lib/runtime") == "/var/lib/runtime"
|
|
monkeypatch.setattr(lease_io.sys, "platform", "darwin")
|
|
assert lease_io._canonical_root("/tmp/runtime") == "/private/tmp/runtime"
|
|
assert lease_io._canonical_root("/var/lib/runtime") == "/private/var/lib/runtime"
|
|
|
|
|
|
def test_read_all_enforces_bound(tmp_path):
|
|
path = tmp_path / "large"
|
|
path.write_bytes(b"0123456789")
|
|
fd = os.open(path, os.O_RDONLY)
|
|
try:
|
|
with pytest.raises(RuntimeError, match="too large"):
|
|
lease_io.read_all(fd, limit=4)
|
|
with pytest.raises(RuntimeError, match="too large"):
|
|
lease_io.read_all(fd, limit=0)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
@pytest.mark.parametrize("failure_stage", ["config-parent", "manifest-parent"])
|
|
def test_retry_reasserts_parent_durability_before_success(tmp_path, monkeypatch, failure_stage):
|
|
events: list[str] = []
|
|
failed = False
|
|
|
|
def fsync(fd: int, stage: str) -> None:
|
|
nonlocal failed
|
|
events.append(stage)
|
|
if stage == failure_stage and not failed:
|
|
failed = True
|
|
raise RuntimeError("injected fsync failure")
|
|
|
|
monkeypatch.setattr(lease_io, "publication_fsync", fsync)
|
|
inp = {
|
|
"data_root": str(tmp_path / "data"), "workspace_id": "abc",
|
|
"workspace_revision": "a" * 40, "config_hex": b"config".hex(),
|
|
"manifest_base": {
|
|
"workspace_id": "abc", "workspace_revision": "a" * 40,
|
|
"descriptor_git_blob": "b" * 40, "descriptor_sha256": "c" * 64,
|
|
"descriptor_dev": "1", "descriptor_ino": "2",
|
|
"config_dwh_binding": {"workspace_id": "abc", "config_fingerprint": "e", "input_fingerprint": "f"},
|
|
},
|
|
}
|
|
with pytest.raises(RuntimeError, match="injected"):
|
|
lease_io.publish(inp)
|
|
events.clear()
|
|
lease_io.publish(inp)
|
|
assert events.index("config-parent") < events.index("manifest-parent")
|
|
|
|
|
|
def test_protocol_success_shapes_and_version_rejection(monkeypatch):
|
|
monkeypatch.setattr(lease_io, "binding", lambda _inp: {
|
|
"workspace_id": "abc", "config_fingerprint": "f", "input_fingerprint": "i",
|
|
})
|
|
monkeypatch.setattr(lease_io.sys, "stdin", io.StringIO(
|
|
'{"protocol_version":1,"action":"binding","config_hex":""}'
|
|
))
|
|
success = io.StringIO()
|
|
monkeypatch.setattr(lease_io.sys, "stdout", success)
|
|
lease_io.main()
|
|
assert set(json.loads(success.getvalue())) == {
|
|
"protocol_version", "kind", "workspace_id", "config_fingerprint", "input_fingerprint",
|
|
}
|
|
|
|
# An old/new protocol mismatch must be rejected before action dispatch.
|
|
monkeypatch.setattr(lease_io.sys, "stdin", io.StringIO(
|
|
'{"protocol_version":999,"action":"binding","config_hex":""}'
|
|
))
|
|
failure = io.StringIO()
|
|
monkeypatch.setattr(lease_io.sys, "stdout", failure)
|
|
with pytest.raises(SystemExit):
|
|
lease_io.main()
|
|
assert json.loads(failure.getvalue())["error"] == "unsupported runtime config protocol"
|
|
|
|
|
|
def test_secure_runtime_reader_binds_path_manifest_and_ignores_legacy_fd(monkeypatch, tmp_path):
|
|
revision = "a" * 40
|
|
inp = {
|
|
"data_root": str(tmp_path / "data"), "workspace_id": "abc",
|
|
"workspace_revision": revision, "config_hex": b"runtime: true\n".hex(),
|
|
"manifest_base": {
|
|
"workspace_id": "abc", "workspace_revision": revision,
|
|
"descriptor_git_blob": "b" * 40, "descriptor_sha256": "c" * 64,
|
|
"descriptor_dev": "1", "descriptor_ino": "2",
|
|
"config_dwh_binding": {"workspace_id": "abc", "config_fingerprint": "e", "input_fingerprint": "f"},
|
|
},
|
|
}
|
|
result = lease_io.publish(inp)
|
|
path = Path(result["path"])
|
|
monkeypatch.setenv("THT_RUNTIME_CONFIG_MANIFEST_SHA256", result["manifest_sha256"])
|
|
monkeypatch.setenv("THT_CONFIG_MANIFEST_FD", "999")
|
|
monkeypatch.setenv("THT_CONFIG_MANIFEST_SHA256", "0" * 64)
|
|
source, manifest = _read_runtime_config_source(path)
|
|
assert source == "runtime: true\n"
|
|
assert manifest is not None and manifest["workspace_id"] == "abc"
|
|
|
|
monkeypatch.setenv("THT_RUNTIME_CONFIG_MANIFEST_SHA256", "0" * 64)
|
|
with pytest.raises(ConfigError):
|
|
_read_runtime_config_source(path)
|
|
with pytest.raises(ConfigError):
|
|
_read_runtime_config_source(path.with_name("not-canonical.yaml"))
|
|
|
|
|
|
@pytest.mark.parametrize("object_kind", ["tree", "blob", "tag"])
|
|
def test_verified_snapshot_rejects_non_commit_object(tmp_path, object_kind):
|
|
repo = tmp_path / "repo"
|
|
(repo / "workspaces").mkdir(parents=True)
|
|
subprocess.run(["git", "init", "--initial-branch=main"], cwd=repo, check=True, stdout=subprocess.DEVNULL)
|
|
subprocess.run(["git", "config", "user.name", "Fixture"], cwd=repo, check=True)
|
|
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], cwd=repo, check=True)
|
|
descriptor = "workspace:\n schema_version: 3\n id: abc\n"
|
|
(repo / "workspaces" / "abc.yaml").write_text(descriptor)
|
|
subprocess.run(["git", "add", "."], cwd=repo, check=True)
|
|
subprocess.run(["git", "commit", "-m", "fixture"], cwd=repo, check=True, stdout=subprocess.DEVNULL)
|
|
if object_kind == "tree":
|
|
revision = subprocess.check_output(["git", "rev-parse", "HEAD^{tree}"], cwd=repo, text=True).strip()
|
|
elif object_kind == "blob":
|
|
revision = subprocess.check_output(["git", "rev-parse", "HEAD:workspaces/abc.yaml"], cwd=repo, text=True).strip()
|
|
else:
|
|
subprocess.run(["git", "tag", "-a", "v1", "-m", "tag"], cwd=repo, check=True)
|
|
revision = subprocess.check_output(["git", "rev-parse", "refs/tags/v1^{tag}"], cwd=repo, text=True).strip()
|
|
|
|
snapshots = tmp_path / "snapshots" / revision
|
|
snapshots.mkdir(parents=True, mode=0o700)
|
|
(tmp_path / "snapshots").chmod(0o700)
|
|
files = {"abc.yaml": descriptor, "abc.env.example": "# fixture\n", "abc.md": "# fixture\n"}
|
|
for name, content in files.items():
|
|
target = snapshots / name
|
|
target.write_text(content)
|
|
target.chmod(0o400)
|
|
snapshot = {
|
|
"head": revision,
|
|
"revisions": [{"id": "abc", "commit": revision, "blob": "b" * 40,
|
|
"snapshotPath": f"{tmp_path / 'snapshots'}/{revision}/abc.yaml"}],
|
|
"files": {name: __import__("hashlib").sha256(content.encode()).hexdigest() for name, content in files.items()},
|
|
}
|
|
manifest = snapshots / "snapshot.json"
|
|
manifest.write_text(json.dumps(snapshot))
|
|
manifest.chmod(0o400)
|
|
with pytest.raises(RuntimeError, match="exact commit|unavailable"):
|
|
lease_io.verified_snapshot({
|
|
"snapshots_root": str(tmp_path / "snapshots"), "repository_root": str(repo),
|
|
"workspace_revision": revision, "workspace_id": "abc",
|
|
})
|