audit-labs/evidence-seal

  1"""Build, load, and verify evidence manifests, and verify seal chains.
  2
  3A manifest is a JSON object describing a sealed directory: every file's SHA-256,
  4a Merkle ``root`` over them, provenance metadata, and a ``previous`` link to an
  5earlier manifest's id. Its own ``id`` is the SHA-256 of its canonical form
  6(excluding ``id`` and ``signature``), so the manifest is self-verifying and can
  7be chained.
  8"""
  9
 10from __future__ import annotations
 11
 12import fnmatch
 13import json
 14from dataclasses import dataclass, field
 15from datetime import datetime, timezone
 16from pathlib import Path
 17
 18from . import ALGORITHM, MANIFEST_VERSION, __version__
 19from .hashing import hash_bytes, hash_file, merkle_root
 20
 21# Keys excluded from the canonical bytes the id is computed over. The id cannot
 22# cover itself; a signature and a timestamp are both applied *to* the id
 23# afterwards, so they sit outside it and compose independently of each other.
 24_ID_EXCLUDED = ("id", "signature", "timestamp")
 25
 26
 27def iter_files(
 28    directory: Path, ignore: list[str] | None, exclude: set[str]
 29) -> list[tuple[str, Path]]:
 30    """Return ``(posix_relpath, abspath)`` for every file under *directory*.
 31
 32    Paths matching an *ignore* glob (against the relative path) or present in
 33    *exclude* are skipped. Results are sorted by relative path.
 34    """
 35    ignore = ignore or []
 36    out: list[tuple[str, Path]] = []
 37    for path in directory.rglob("*"):
 38        if not path.is_file():
 39            continue
 40        rel = path.relative_to(directory).as_posix()
 41        if rel in exclude or any(fnmatch.fnmatch(rel, pat) for pat in ignore):
 42            continue
 43        out.append((rel, path))
 44    out.sort(key=lambda pair: pair[0])
 45    return out
 46
 47
 48def canonical_bytes(manifest: dict, exclude: tuple[str, ...] = _ID_EXCLUDED) -> bytes:
 49    """Serialize a manifest deterministically for hashing or signing."""
 50    trimmed = {k: v for k, v in manifest.items() if k not in exclude}
 51    return json.dumps(trimmed, sort_keys=True, separators=(",", ":")).encode()
 52
 53
 54def compute_id(manifest: dict) -> str:
 55    """The manifest's self-id: SHA-256 of its canonical form."""
 56    return hash_bytes(canonical_bytes(manifest))
 57
 58
 59def build_manifest(
 60    directory: str | Path,
 61    metadata: dict | None = None,
 62    previous: str | None = None,
 63    ignore: list[str] | None = None,
 64    exclude: set[str] | None = None,
 65) -> dict:
 66    """Seal *directory* into a manifest dict (id included, unsigned)."""
 67    directory = Path(directory)
 68    if not directory.is_dir():
 69        raise FileNotFoundError(f"not a directory: {directory}")
 70
 71    files = iter_files(directory, ignore, exclude or set())
 72    entries = [
 73        {"path": rel, ALGORITHM: hash_file(abs_path), "bytes": abs_path.stat().st_size}
 74        for rel, abs_path in files
 75    ]
 76    hashes = {entry["path"]: entry[ALGORITHM] for entry in entries}
 77
 78    manifest = {
 79        "manifest_version": MANIFEST_VERSION,
 80        "algorithm": ALGORITHM,
 81        "tool": f"evidence-seal/{__version__}",
 82        "created_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
 83        "subject": directory.name,
 84        "previous": previous,
 85        "metadata": dict(sorted((metadata or {}).items())),
 86        "ignore": sorted(ignore or []),
 87        "file_count": len(entries),
 88        "total_bytes": sum(entry["bytes"] for entry in entries),
 89        "root": merkle_root(hashes),
 90        "files": entries,
 91    }
 92    manifest["id"] = compute_id(manifest)
 93    return manifest
 94
 95
 96def write_manifest(manifest: dict, path: str | Path) -> None:
 97    """Write a manifest as pretty JSON (stable key order)."""
 98    Path(path).write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
 99
100
101def load_manifest(path: str | Path) -> dict:
102    """Load a manifest JSON file."""
103    return json.loads(Path(path).read_text(encoding="utf-8"))
104
105
106@dataclass
107class VerifyResult:
108    """The outcome of verifying a directory against its manifest."""
109
110    modified: list[str] = field(default_factory=list)
111    added: list[str] = field(default_factory=list)
112    removed: list[str] = field(default_factory=list)
113    id_ok: bool = True
114    root_ok: bool = True
115    checked: int = 0
116
117    @property
118    def intact(self) -> bool:
119        """True only if nothing drifted and the manifest is internally sound."""
120        return (
121            not self.modified
122            and not self.added
123            and not self.removed
124            and self.id_ok
125            and self.root_ok
126        )
127
128
129def verify_manifest(
130    directory: str | Path,
131    manifest: dict,
132    ignore: list[str] | None = None,
133    exclude: set[str] | None = None,
134) -> VerifyResult:
135    """Compare the current contents of *directory* against *manifest*."""
136    directory = Path(directory)
137    recorded = {entry["path"]: entry[ALGORITHM] for entry in manifest.get("files", [])}
138
139    result = VerifyResult()
140    # The manifest is only trustworthy if its id and root re-derive.
141    result.id_ok = compute_id(manifest) == manifest.get("id")
142    result.root_ok = merkle_root(recorded) == manifest.get("root")
143
144    present: dict[str, str] = {}
145    for rel, abs_path in iter_files(directory, ignore, exclude or set()):
146        present[rel] = hash_file(abs_path)
147
148    result.checked = len(present)
149    for rel, digest in present.items():
150        if rel not in recorded:
151            result.added.append(rel)
152        elif recorded[rel] != digest:
153            result.modified.append(rel)
154    result.removed = [rel for rel in recorded if rel not in present]
155
156    for bucket in (result.modified, result.added, result.removed):
157        bucket.sort()
158    return result
159
160
161@dataclass
162class ChainResult:
163    """The outcome of verifying a sequence of chained manifests."""
164
165    ok: bool
166    length: int
167    broken_at: int | None = None  # index whose `previous` did not match
168    reason: str = ""
169
170
171def verify_chain(manifests: list[dict]) -> ChainResult:
172    """Verify manifests link oldest→newest via ``previous`` == prior ``id``.
173
174    Also checks that each manifest's own id re-derives, so a tampered link in
175    the middle is caught whether the break is in the pointer or the content.
176    """
177    if not manifests:
178        return ChainResult(ok=False, length=0, reason="no manifests given")
179
180    for index, manifest in enumerate(manifests):
181        if compute_id(manifest) != manifest.get("id"):
182            return ChainResult(False, len(manifests), index, "manifest id does not re-derive")
183        expected_prev = manifests[index - 1]["id"] if index else None
184        if manifest.get("previous") != expected_prev:
185            reason = (
186                "first manifest should have no previous"
187                if index == 0
188                else "previous does not match the prior manifest id"
189            )
190            return ChainResult(False, len(manifests), index, reason)
191
192    return ChainResult(ok=True, length=len(manifests))