#!/usr/bin/env python3 """Verify a DFIRe chain-of-custody export. Runs on Python 3, no dependencies. python3 verify_custody_chain.py custody-chain-.json Checks that every entry's hash matches its contents, that each entry links to the one before it, that the first links to a genesis derived from the installation and item, and that sequence numbers run 1..N with no gaps. Cannot detect entries removed from the end of a chain: a shortened chain still verifies. Compare an entry number and hash against an earlier copy, such as a printed receipt. Nor does it say whether a recorded handover really happened. Exit status: 0 verified, 1 not verified, 2 usage error. https://dfire.fi/docs/evidence.html#chain-of-custody """ import hashlib import json import sys GENESIS_DOMAIN = b"DFIRe custody genesis v1\x00" ENTRY_DOMAIN = b"DFIRe custody entry v1\x00" SUPPORTED_HASH_VERSION = 1 ENTRY_SCHEMA = "dfire.custody.entry.v1" # Exactly the keys a version 1 entry carries, in sorted order. ENTRY_KEYS = [ "condition", "from_party", "item_uuid", "location", "notes", "previous_hash", "purpose", "recorded_at", "recorded_by", "schema", "sequence", "tenant_uuid", "to_party", "transfer_datetime", "transfer_type", ] def canonical_bytes(entry): """Serialize an entry exactly as DFIRe hashed it: sorted keys, no whitespace, UTF-8. Numbers never appear; sequences are decimal strings.""" return json.dumps( entry, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False, ).encode("utf-8") def genesis_hash(tenant_uuid, item_uuid): payload = f"{tenant_uuid.lower()}/{item_uuid.lower()}".encode() return hashlib.sha256(GENESIS_DOMAIN + payload).hexdigest() def entry_hash(entry): return hashlib.sha256(ENTRY_DOMAIN + canonical_bytes(entry)).hexdigest() def verify(document): """Return (ok, list of human-readable lines).""" lines = [] tenant_uuid = document["tenant_uuid"] item_uuid = document["item_uuid"] expected_previous = genesis_hash(tenant_uuid, item_uuid) lines.append(f"Case: {document.get('case_number', '(unknown)')}") lines.append(f"Evidence: {document.get('item_name') or item_uuid}") lines.append(f"Item: {item_uuid}") lines.append(f"Genesis: {expected_previous}") if expected_previous != document.get("genesis_hash"): lines.append("") lines.append( "FAILED: the genesis hash in the file does not match the one derived" ) lines.append( " from its tenant and item UUID. The file has been altered, or" ) lines.append(" it came from a different installation or item.") return False, lines lines.append("") entries = document.get("entries", []) # Before the empty case returns: a file claiming a count and head while # listing no entries is the shape a doctored export takes. declared_count = document.get("entry_count") if declared_count is not None and str(declared_count) != str(len(entries)): lines.append( f"FAILED: the file says it holds {declared_count} entries but " f"carries {len(entries)}." ) return False, lines if not entries: declared_head = document.get("head_hash") if declared_head and declared_head != expected_previous: lines.append("FAILED: the file carries no entries, but states a chain") lines.append(" hash other than the empty chain's starting value.") return False, lines lines.append("This item has no custody entries.") return True, lines for position, record in enumerate(entries, start=1): entry = record["canonical_entry"] recorded = record["entry_hash"] version = record.get("hash_version") # Reject anything that is not exactly a version 1 entry: an added or # missing key would be hashed as-is and agree with its own digest. if sorted(entry) != ENTRY_KEYS: lines.append(f"FAILED at entry #{position}: unexpected fields.") lines.append(f" expected {ENTRY_KEYS}") lines.append(f" found {sorted(entry)}") return False, lines if entry.get("schema") != ENTRY_SCHEMA: lines.append( f"FAILED at entry #{position}: schema is " f"{entry.get('schema')!r}, expected {ENTRY_SCHEMA!r}." ) return False, lines if ( entry.get("tenant_uuid") != tenant_uuid or entry.get("item_uuid") != item_uuid ): lines.append( f"FAILED at entry #{position}: belongs to a different " f"installation or evidence item than this file claims." ) return False, lines if version != SUPPORTED_HASH_VERSION: lines.append( f"FAILED at entry #{position}: unsupported hash version {version}." ) lines.append(" This script understands version 1 only.") return False, lines if entry.get("sequence") != str(position): lines.append( f"FAILED at entry #{position}: sequence is {entry.get('sequence')!r}, " f"expected {str(position)!r}." ) lines.append(" An entry is missing or duplicated.") return False, lines if entry.get("previous_hash") != expected_previous: lines.append( f"FAILED at entry #{position}: does not link to the entry before it." ) lines.append(f" expected previous {expected_previous}") lines.append(f" found previous {entry.get('previous_hash')}") return False, lines recomputed = entry_hash(entry) if recomputed != recorded: lines.append( f"FAILED at entry #{position}: contents do not match the recorded hash." ) lines.append(f" recorded {recorded}") lines.append(f" recomputed {recomputed}") lines.append(" Some field of this entry has been altered.") return False, lines who = (entry.get("to_party") or {}).get("display_name", "?") lines.append( f" #{position} {entry.get('transfer_type'):<10} " f"{entry.get('transfer_datetime')} -> {who}" ) lines.append(f" {recorded}") expected_previous = recorded declared_head = document.get("head_hash") if declared_head and declared_head != expected_previous: lines.append("") lines.append("FAILED: the file's stated chain hash is not the hash of its") lines.append(" last entry.") lines.append(f" stated {declared_head}") lines.append(f" recomputed {expected_previous}") return False, lines lines.append("") lines.append(f"{len(entries)} entries verified.") lines.append(f"Head hash: {expected_previous}") lines.append("") lines.append("To detect removals from the end, compare an entry number and hash") lines.append("above against an earlier copy. A longer chain is normal; a shorter") lines.append("one, or a changed hash at that number, is not.") return True, lines def main(argv): if len(argv) != 2: print(__doc__) return 2 try: with open(argv[1], encoding="utf-8") as handle: document = json.load(handle) except OSError as exc: print(f"Could not read {argv[1]}: {exc}") return 2 except json.JSONDecodeError as exc: print(f"{argv[1]} is not valid JSON: {exc}") return 2 if document.get("schema") != "dfire.custody.chain_export.v1": print(f"Unexpected document type: {document.get('schema')!r}") print("This script reads a DFIRe custody chain export, schema version 1.") return 2 ok, lines = verify(document) for line in lines: print(line) return 0 if ok else 1 if __name__ == "__main__": sys.exit(main(sys.argv))