diff --git a/scripts/leo_identity_version_manifest.py b/scripts/leo_identity_version_manifest.py new file mode 100644 index 0000000..f2fac26 --- /dev/null +++ b/scripts/leo_identity_version_manifest.py @@ -0,0 +1,422 @@ +#!/usr/bin/env python3 +"""Generate and independently rebuild Leo's deterministic identity version manifest. + +The version manifest is an operator-facing projection over the stricter identity +compiler. It names the runtime, Git commit, individual skills, database +version, compiled identity, and session boundary without turning generated +views or session history into provenance. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from pathlib import Path, PurePosixPath +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from scripts import leo_behavior_manifest as behavior +from scripts import leo_identity_manifest as identity + +SCHEMA = "livingip.leoIdentityVersionManifest.v1" +DEFAULT_PROFILE = Path("fixtures/working-leo/leo-identity-v1/profile-template") +DEFAULT_HERMES_ROOT = Path("fixtures/working-leo/leo-identity-v1/hermes-runtime") +DEFAULT_DEPLOYMENT_ROOT = Path(".") +DEFAULT_SOURCE_ROOT = Path(".") +DEFAULT_DATABASE_FINGERPRINT = Path("fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json") +DEFAULT_CONSTITUTION = Path("fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json") +DEFAULT_DATABASE_IDENTITY = Path("fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json") +HEX_64 = re.compile(r"^[0-9a-f]{64}$") + + +class IdentityVersionManifestError(identity.IdentityManifestError): + """The identity version receipt is incomplete, inconsistent, or drifted.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise IdentityVersionManifestError(message) + + +def expected_authority_contract() -> dict[str, str]: + return { + "durable_knowledge_authority": "approved_postgres_rows_with_bound_evidence", + "compiled_identity_role": "derived_view_from_versioned_inputs_not_authority", + "session_history_role": "temporary_or_persisted_context_not_provenance", + } + + +def _runtime_version(identity_manifest: dict[str, Any]) -> dict[str, Any]: + core = identity_manifest["identity_inputs"] + runtime = core["runtime"] + model = core["model"] + return { + "source_commit": core["source_commit"], + "identity_runtime_sha256": runtime["identity_runtime_sha256"], + "teleo_infrastructure": { + "git_head": runtime["teleo_git_head"], + "identity_source_sha256": runtime["teleo_source_sha256"], + }, + "hermes": { + "git_head": runtime["hermes_git_head"], + "source_sha256": runtime["hermes_source_sha256"], + }, + "python": runtime["python"], + "model": { + "provider": model["provider"], + "default_model": model["default_model"], + "model_section_sha256": model["model_section_sha256"], + "hosted_weights": model["hosted_weights"], + "actual_model_required_in_turn_receipt": model["actual_model_required_in_turn_receipt"], + }, + } + + +def _canonical_database_version(identity_manifest: dict[str, Any]) -> dict[str, Any]: + core = identity_manifest["identity_inputs"] + database = core["canonical_database"] + source = core["identity_sources"]["database_derived_identity"] + provenance = source["provenance"] + return { + "authority": database["authority"], + "canonical_authority_granted": database["canonical_authority_granted"], + "database": database["database"], + "database_user": database["database_user"], + "system_identifier": database["system_identifier"], + "fingerprint_sha256": database["fingerprint_sha256"], + "table_rows_sha256": database["table_rows_sha256"], + "structure_sha256": database["structure_sha256"], + "table_count": database["table_count"], + "total_rows": database["total_rows"], + "identity_records_sha256": source["semantic_sha256"], + "identity_record_count": source["record_count"], + "source_mode": provenance["mode"], + "same_transaction_as_fingerprint": provenance["same_transaction_as_fingerprint"], + "export_receipt_sha256": provenance["export_receipt_sha256"], + } + + +def _compiled_identity_version(identity_manifest: dict[str, Any]) -> dict[str, Any]: + return { + "role": "derived_view_from_versioned_inputs_not_authority", + "identity_inputs_sha256": identity_manifest["identity_inputs_sha256"], + "identity_manifest_sha256": identity_manifest["manifest_sha256"], + "views": identity_manifest["compiled_views"], + } + + +def _skill_versions(behavior_manifest: dict[str, Any], identity_manifest: dict[str, Any]) -> dict[str, Any]: + content = behavior_manifest["components"]["procedural_skills"]["content"] + identity.validate_content_manifest(content, label="procedural skills") + groups: dict[str, list[dict[str, Any]]] = {} + for item in content["files"]: + parts = PurePosixPath(item["path"]).parts + _require( + len(parts) >= 3 and parts[0] == "skills" and bool(parts[1]), + "procedural skill file is not inside a named skill directory", + ) + relative_path = PurePosixPath(*parts[2:]).as_posix() + groups.setdefault(parts[1], []).append( + { + "path": relative_path, + "bytes": item["bytes"], + "mode": item["mode"], + "sha256": item["sha256"], + } + ) + _require(bool(groups), "at least one named skill is required") + skills = [] + for name in sorted(groups): + files = sorted(groups[name], key=lambda item: item["path"]) + skills.append( + { + "name": name, + "files": files, + "file_count": len(files), + "total_bytes": sum(item["bytes"] for item in files), + "content_sha256": behavior.canonical_sha256(files), + } + ) + result = { + "content_sha256": content["sha256"], + "file_count": content["file_count"], + "total_bytes": content["total_bytes"], + "skills": skills, + } + expected = identity_manifest["identity_inputs"]["skills"] + _require(result["content_sha256"] == expected["content_sha256"], "skill aggregate hash drift detected") + _require(result["file_count"] == expected["file_count"], "skill aggregate file count drift detected") + return result + + +def _validate_skill_versions(value: Any, identity_manifest: dict[str, Any]) -> None: + _require(isinstance(value, dict), "skill versions are missing") + _require( + set(value) == {"content_sha256", "file_count", "total_bytes", "skills"}, + "skill version fields are invalid", + ) + _require(isinstance(value.get("skills"), list) and bool(value["skills"]), "named skills are missing") + names: list[str] = [] + reconstructed_files: list[dict[str, Any]] = [] + total_bytes = 0 + for skill in value["skills"]: + _require( + isinstance(skill, dict) and set(skill) == {"name", "files", "file_count", "total_bytes", "content_sha256"}, + "named skill fields are invalid", + ) + name = skill.get("name") + files = skill.get("files") + _require( + isinstance(name, str) and bool(name) and PurePosixPath(name).parts == (name,) and name not in {".", ".."}, + "skill name is invalid", + ) + _require(isinstance(files, list) and bool(files), f"skill {name} has no files") + names.append(name) + paths: list[str] = [] + skill_bytes = 0 + for item in files: + _require( + isinstance(item, dict) and set(item) == {"path", "bytes", "mode", "sha256"}, + f"skill {name} file fields are invalid", + ) + path = item.get("path") + size = item.get("bytes") + mode = item.get("mode") + _require( + isinstance(path, str) + and bool(path) + and not path.startswith("/") + and ".." not in PurePosixPath(path).parts, + f"skill {name} path is invalid", + ) + _require(isinstance(size, int) and not isinstance(size, bool) and size >= 0, "skill byte size is invalid") + _require(isinstance(mode, str) and bool(re.fullmatch(r"0o[0-7]{3}", mode)), "skill mode is invalid") + _require( + isinstance(item.get("sha256"), str) and bool(HEX_64.fullmatch(item["sha256"])), "skill hash is invalid" + ) + paths.append(path) + skill_bytes += size + reconstructed_files.append({**item, "path": f"skills/{name}/{path}"}) + _require(paths == sorted(paths) and len(paths) == len(set(paths)), f"skill {name} paths are not sorted") + _require(skill.get("file_count") == len(files), f"skill {name} file count is invalid") + _require(skill.get("total_bytes") == skill_bytes, f"skill {name} byte count is invalid") + _require(skill.get("content_sha256") == behavior.canonical_sha256(files), f"skill {name} hash drift detected") + total_bytes += skill_bytes + _require(names == sorted(names) and len(names) == len(set(names)), "skill names are not unique and sorted") + reconstructed_files.sort(key=lambda item: item["path"]) + aggregate = behavior.canonical_sha256({"files": reconstructed_files, "missing": [], "symlinks": []}) + _require(value.get("content_sha256") == aggregate, "skill aggregate content drift detected") + _require(value.get("file_count") == len(reconstructed_files), "skill aggregate file count is invalid") + _require(value.get("total_bytes") == total_bytes, "skill aggregate byte count is invalid") + expected = identity_manifest["identity_inputs"]["skills"] + _require(value["content_sha256"] == expected["content_sha256"], "skills are not bound to the identity manifest") + _require(value["file_count"] == expected["file_count"], "skill count is not bound to the identity manifest") + + +def assemble_version_manifest( + *, behavior_manifest: dict[str, Any], identity_manifest: dict[str, Any] +) -> dict[str, Any]: + identity.validate_behavior_manifest(behavior_manifest) + identity.validate_identity_manifest(identity_manifest) + core = identity_manifest["identity_inputs"] + _require( + behavior_manifest["identity_runtime_sha256"] == core["runtime"]["identity_runtime_sha256"], + "behavior runtime is not bound to the identity manifest", + ) + _require( + behavior_manifest["teleo_infrastructure_runtime"]["git_head"] == core["source_commit"], + "behavior source commit is not bound to the identity manifest", + ) + stable = { + "schema": SCHEMA, + "identity_name": "Leo", + "runtime_version": _runtime_version(identity_manifest), + "skill_versions": _skill_versions(behavior_manifest, identity_manifest), + "canonical_database_version": _canonical_database_version(identity_manifest), + "compiled_identity_version": _compiled_identity_version(identity_manifest), + "session_boundary": core["session_boundary"], + "authority_contract": expected_authority_contract(), + "identity_manifest": identity_manifest, + } + return {**stable, "manifest_sha256": behavior.canonical_sha256(stable)} + + +def build_version_manifest( + *, + profile: Path, + hermes_root: Path, + deployment_root: Path, + source_root: Path, + database_fingerprint: Path, + constitution: Path, + database_identity: Path, +) -> dict[str, Any]: + behavior_manifest = behavior.build_manifest(profile, hermes_root, deployment_root) + identity_manifest = identity.build_identity_manifest( + behavior_manifest=behavior_manifest, + database_fingerprint_path=database_fingerprint, + constitution_path=constitution, + database_identity_path=database_identity, + source_root=source_root, + ) + identity.verify_bound_sources(identity_manifest, source_root) + return assemble_version_manifest( + behavior_manifest=behavior_manifest, + identity_manifest=identity_manifest, + ) + + +def validate_version_manifest(value: dict[str, Any]) -> None: + _require( + set(value) + == { + "schema", + "identity_name", + "runtime_version", + "skill_versions", + "canonical_database_version", + "compiled_identity_version", + "session_boundary", + "authority_contract", + "identity_manifest", + "manifest_sha256", + }, + "identity version manifest fields are invalid", + ) + _require(value.get("schema") == SCHEMA, "unsupported identity version manifest schema") + _require(value.get("identity_name") == "Leo", "identity version must describe Leo") + expected_hash = value.get("manifest_sha256") + _require(isinstance(expected_hash, str) and bool(HEX_64.fullmatch(expected_hash)), "manifest hash is invalid") + stable = {key: item for key, item in value.items() if key != "manifest_sha256"} + _require(behavior.canonical_sha256(stable) == expected_hash, "identity version manifest hash drift detected") + embedded = value.get("identity_manifest") + _require(isinstance(embedded, dict), "embedded identity manifest is missing") + identity.validate_identity_manifest(embedded) + core = embedded["identity_inputs"] + _require(value.get("runtime_version") == _runtime_version(embedded), "runtime version projection drift detected") + _validate_skill_versions(value.get("skill_versions"), embedded) + _require( + value.get("canonical_database_version") == _canonical_database_version(embedded), + "canonical database version projection drift detected", + ) + _require( + value.get("compiled_identity_version") == _compiled_identity_version(embedded), + "compiled identity version projection drift detected", + ) + _require(value.get("session_boundary") == core["session_boundary"], "session boundary projection drift detected") + _require( + value.get("authority_contract") == expected_authority_contract(), "identity authority contract drift detected" + ) + + +def verify_against_inputs( + value: dict[str, Any], + *, + profile: Path, + hermes_root: Path, + deployment_root: Path, + source_root: Path, + database_fingerprint: Path, + constitution: Path, + database_identity: Path, +) -> None: + validate_version_manifest(value) + rebuilt = build_version_manifest( + profile=profile, + hermes_root=hermes_root, + deployment_root=deployment_root, + source_root=source_root, + database_fingerprint=database_fingerprint, + constitution=constitution, + database_identity=database_identity, + ) + _require(rebuilt == value, "identity version manifest does not match the supplied versioned inputs") + + +def write_json(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def summary(value: dict[str, Any]) -> dict[str, Any]: + return { + "status": "ok", + "manifest_sha256": value["manifest_sha256"], + "source_commit": value["runtime_version"]["source_commit"], + "skills": [item["name"] for item in value["skill_versions"]["skills"]], + "canonical_database": { + "authority": value["canonical_database_version"]["authority"], + "canonical_authority_granted": value["canonical_database_version"]["canonical_authority_granted"], + "fingerprint_sha256": value["canonical_database_version"]["fingerprint_sha256"], + }, + "compiled_identity_sha256": value["compiled_identity_version"]["identity_inputs_sha256"], + "session_boundary": value["session_boundary"]["classification"], + } + + +def _add_input_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--profile", type=Path, default=DEFAULT_PROFILE) + parser.add_argument("--hermes-root", type=Path, default=DEFAULT_HERMES_ROOT) + parser.add_argument("--deployment-root", type=Path, default=DEFAULT_DEPLOYMENT_ROOT) + parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT) + parser.add_argument("--database-fingerprint", type=Path, default=DEFAULT_DATABASE_FINGERPRINT) + parser.add_argument("--constitution", type=Path, default=DEFAULT_CONSTITUTION) + parser.add_argument("--database-identity", type=Path, default=DEFAULT_DATABASE_IDENTITY) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + generate = subparsers.add_parser("generate", help="generate a deterministic identity version manifest") + _add_input_arguments(generate) + generate.add_argument("--output", type=Path, required=True) + verify = subparsers.add_parser("verify", help="rebuild and compare a manifest from its versioned inputs") + verify.add_argument("--manifest", type=Path, required=True) + _add_input_arguments(verify) + args = parser.parse_args() + try: + if args.command == "generate": + manifest = build_version_manifest( + profile=args.profile, + hermes_root=args.hermes_root, + deployment_root=args.deployment_root, + source_root=args.source_root, + database_fingerprint=args.database_fingerprint, + constitution=args.constitution, + database_identity=args.database_identity, + ) + write_json(args.output, manifest) + else: + manifest = identity.load_json(args.manifest, "identity version manifest") + verify_against_inputs( + manifest, + profile=args.profile, + hermes_root=args.hermes_root, + deployment_root=args.deployment_root, + source_root=args.source_root, + database_fingerprint=args.database_fingerprint, + constitution=args.constitution, + database_identity=args.database_identity, + ) + print(json.dumps(summary(manifest), sort_keys=True)) + return 0 + except (identity.IdentityManifestError, OSError, ValueError) as exc: + print( + json.dumps( + { + "status": "error", + "error": "identity_version_manifest_invalid", + "message": str(exc), + }, + sort_keys=True, + ) + ) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_leo_identity_version_manifest.py b/tests/test_leo_identity_version_manifest.py new file mode 100644 index 0000000..f610622 --- /dev/null +++ b/tests/test_leo_identity_version_manifest.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from scripts import leo_behavior_manifest as behavior +from scripts import leo_identity_version_manifest as version +from tests.test_leo_identity_manifest import _fixture, _write + +ROOT = Path(__file__).resolve().parents[1] + + +def _build(fixture: dict[str, Path | dict]) -> dict: + return version.build_version_manifest( + profile=fixture["profile"], + hermes_root=fixture["hermes"], + deployment_root=fixture["repo"], + source_root=fixture["repo"], + database_fingerprint=fixture["fingerprint"], + constitution=fixture["constitution"], + database_identity=fixture["database_identity"], + ) + + +def _rehash(value: dict) -> None: + stable = {key: item for key, item in value.items() if key != "manifest_sha256"} + value["manifest_sha256"] = behavior.canonical_sha256(stable) + + +def _input_args(fixture: dict[str, Path | dict]) -> list[str]: + return [ + "--profile", + str(fixture["profile"]), + "--hermes-root", + str(fixture["hermes"]), + "--deployment-root", + str(fixture["repo"]), + "--source-root", + str(fixture["repo"]), + "--database-fingerprint", + str(fixture["fingerprint"]), + "--constitution", + str(fixture["constitution"]), + "--database-identity", + str(fixture["database_identity"]), + ] + + +def test_version_manifest_names_every_required_identity_boundary(tmp_path: Path) -> None: + manifest = _build(_fixture(tmp_path)) + + assert manifest["schema"] == version.SCHEMA + assert ( + manifest["runtime_version"]["source_commit"] == manifest["runtime_version"]["teleo_infrastructure"]["git_head"] + ) + assert [item["name"] for item in manifest["skill_versions"]["skills"]] == ["identity"] + assert manifest["canonical_database_version"] == { + "authority": "synthetic_noncanonical_fixture", + "canonical_authority_granted": False, + "database": "fixture", + "database_user": "reader", + "system_identifier": "12345", + "fingerprint_sha256": "1" * 64, + "table_rows_sha256": "2" * 64, + "structure_sha256": "3" * 64, + "table_count": 2, + "total_rows": 2, + "identity_records_sha256": manifest["identity_manifest"]["identity_inputs"]["identity_sources"][ + "database_derived_identity" + ]["semantic_sha256"], + "identity_record_count": 2, + "source_mode": "synthetic_noncanonical_fixture", + "same_transaction_as_fingerprint": False, + "export_receipt_sha256": None, + } + assert manifest["compiled_identity_version"]["role"] == "derived_view_from_versioned_inputs_not_authority" + assert manifest["session_boundary"]["included_in_identity_inputs"] is False + assert manifest["session_boundary"]["may_be_used_as_evidence"] is False + assert manifest["authority_contract"] == version.expected_authority_contract() + version.validate_version_manifest(manifest) + + +def test_version_manifest_is_identical_across_checkout_paths(tmp_path: Path) -> None: + first = _build(_fixture(tmp_path / "checkout-a")) + second = _build(_fixture(tmp_path / "checkout-b")) + + assert first == second + assert str(tmp_path) not in json.dumps(first, sort_keys=True) + + +@pytest.mark.parametrize( + ("field", "replacement", "message"), + [ + ( + "authority_contract", + { + "durable_knowledge_authority": "session_history", + "compiled_identity_role": "authority", + "session_history_role": "provenance", + }, + "authority contract drift", + ), + ( + "session_boundary", + { + "classification": "canonical", + "included_in_identity_inputs": True, + "may_be_used_as_evidence": True, + "restart_policy": "reuse_history_as_identity", + }, + "session boundary projection drift", + ), + ], +) +def test_rehashed_authority_or_session_tampering_fails_closed( + tmp_path: Path, field: str, replacement: dict, message: str +) -> None: + manifest = _build(_fixture(tmp_path)) + manifest[field] = replacement + _rehash(manifest) + + with pytest.raises(version.IdentityVersionManifestError, match=message): + version.validate_version_manifest(manifest) + + +def test_rehashed_database_authority_tampering_fails_closed(tmp_path: Path) -> None: + manifest = _build(_fixture(tmp_path)) + manifest["canonical_database_version"]["canonical_authority_granted"] = True + manifest["canonical_database_version"]["authority"] = "canonical_database" + _rehash(manifest) + + with pytest.raises(version.IdentityVersionManifestError, match="canonical database version projection drift"): + version.validate_version_manifest(manifest) + + +def test_named_skill_manifest_rejects_fully_rehashed_path_or_hash_tampering(tmp_path: Path) -> None: + manifest = _build(_fixture(tmp_path)) + skill = manifest["skill_versions"]["skills"][0] + skill["files"][0]["path"] = "../MEMORY.md" + skill["content_sha256"] = behavior.canonical_sha256(skill["files"]) + _rehash(manifest) + + with pytest.raises(version.IdentityVersionManifestError, match="skill identity path is invalid"): + version.validate_version_manifest(manifest) + + +def test_named_skill_manifest_rejects_a_rehashed_parent_name(tmp_path: Path) -> None: + manifest = _build(_fixture(tmp_path)) + manifest["skill_versions"]["skills"][0]["name"] = ".." + _rehash(manifest) + + with pytest.raises(version.IdentityVersionManifestError, match="skill name is invalid"): + version.validate_version_manifest(manifest) + + +def test_rebuilder_rejects_a_different_clean_input_commit(tmp_path: Path) -> None: + fixture = _fixture(tmp_path) + manifest = _build(fixture) + _write(fixture["profile"] / "skills" / "identity" / "SKILL.md", "# changed identity readback\n") + subprocess.run(["git", "-C", str(fixture["repo"]), "add", "."], check=True) + subprocess.run( + ["git", "-C", str(fixture["repo"]), "commit", "-qm", "change identity input"], + check=True, + env={ + **dict(os.environ), + "GIT_AUTHOR_DATE": "2026-01-02T00:00:00Z", + "GIT_COMMITTER_DATE": "2026-01-02T00:00:00Z", + }, + ) + + with pytest.raises(version.IdentityVersionManifestError, match="does not match the supplied versioned inputs"): + version.verify_against_inputs( + manifest, + profile=fixture["profile"], + hermes_root=fixture["hermes"], + deployment_root=fixture["repo"], + source_root=fixture["repo"], + database_fingerprint=fixture["fingerprint"], + constitution=fixture["constitution"], + database_identity=fixture["database_identity"], + ) + + +def test_cli_generate_is_deterministic_and_verify_rebuilds_inputs(tmp_path: Path) -> None: + fixture = _fixture(tmp_path) + first = tmp_path / "identity-version-first.json" + second = tmp_path / "identity-version-second.json" + common = _input_args(fixture) + + for output in (first, second): + result = subprocess.run( + [ + sys.executable, + "-m", + "scripts.leo_identity_version_manifest", + "generate", + *common, + "--output", + str(output), + ], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ) + summary = json.loads(result.stdout) + assert summary["status"] == "ok" + assert summary["skills"] == ["identity"] + + assert first.read_bytes() == second.read_bytes() + verify = subprocess.run( + [ + sys.executable, + "-m", + "scripts.leo_identity_version_manifest", + "verify", + "--manifest", + str(first), + *common, + ], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ) + assert json.loads(verify.stdout)["manifest_sha256"] == json.loads(first.read_text())["manifest_sha256"]