fix: harden runtime config helper protocol and lifecycle

harness/tests/test_runtime_config_lease_io.py

@@ -1,10 +1,15 @@
 """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:
@@ -82,3 +87,99 @@ def test_retry_reasserts_parent_durability_before_success(tmp_path, monkeypatch,
     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",
+        })