How to Build a Tamper-Evident Audit Log for AI Agent Actions in Python With Hash Chains, Signed Checkpoints, and Key Evolution

An AI agent that reads files, runs commands and calls APIs leaves one thing behind that you can examine afterwards: its log. If anyone who can write to the log file can also rewrite it, whether that is a bug, the agent itself or an attacker who has taken over the host, the log stops being evidence and becomes a story. In this tutorial you will build an audit log for agent actions that is tamper-evident. It cannot stop someone with write access from changing it, but every edit, deletion and reordering shows up when you verify it, and you will see exactly which attacks the design catches and which ones it cannot.

You will start with an ordinary JSON Lines log and watch an attacker erase two entries without leaving a trace. Then you will add a hash chain, break it, add a key, break that, make the key change after every entry, and finally add signed checkpoints that live somewhere else. Each layer exists because the previous one failed in a way you will reproduce yourself.

The question is current. Our look at Nadella’s insider-risk essay found that NVIDIA’s OpenShell documentation says its file log writers can drop events under pressure, while the essay itself asked for tamper-proof, human-readable evidence. And the agent harness tutorial writes a JSON Lines event log; this tutorial is what you add so that kind of log can be trusted.

What tamper-evident means, and the words you need first

Tamper-evident is a weaker promise than tamper-proof, and it is the honest one. A tamper-proof log would make changes impossible, which no software on a machine an attacker controls can do. A tamper-evident log lets anyone who holds the right secret or reference value check, later, whether the log is still what it was. A few terms, defined before you meet them in code:

• Hash. A fixed-size fingerprint of some bytes, here SHA-256, which gives 64 hexadecimal characters. Change one bit of the input and the fingerprint changes completely, and nobody can work backwards from a fingerprint to the input.
• Hash chain. Each log entry stores the hash of the entry before it, and its own hash covers that stored value. Changing an old entry changes its hash, which no longer matches what the next entry recorded.
• MAC (message authentication code). A fingerprint that needs a secret key to produce, here HMAC-SHA256. Anyone can compute a plain hash; only a key holder can compute a MAC.
• Key evolution, or forward security. The key changes after every entry by running it through a one-way function, and the old key is overwritten. Someone who steals today’s key cannot turn it back into yesterday’s.
• Digital signature. Like a MAC, except the key that checks it (the public key) cannot be used to create it. This tutorial uses Ed25519 from the cryptography package.
• Checkpoint and anchor. A signed statement of the form the log had entry number N and its hash was H, stored somewhere the log host cannot reach. Storing it is called anchoring.

What you will build, and what you will not

By the end you will have a small Python package that:

• writes agent actions to a JSON Lines file as a hash chain, with a strict reader that refuses anything but the exact bytes the writer produced;
• adds an HMAC to every entry and changes the key after every entry;
• signs checkpoints and checks a log against them;
• stores the log in SQLite so several writer processes take turns instead of forking the chain;
• wraps an agent’s tools so each call is logged before it runs;
• comes with 54 tests and a script that plants 12 bugs to prove the tests notice.

Be clear about the limits before you start. A log can only prove that records were not changed after they were written. It cannot prove the records were true: an agent that is already compromised can log lies, and an action that never reached the logger leaves no entry to protect. The code here has one writer process that holds the key in memory, and Python cannot reliably wipe a secret from memory. It is a teaching implementation of a well-studied design, not a hardened product.

Prerequisites

• Python 3.13. Everything here ran on Windows 11 with Python 3.13.14, and I did not test older versions. The code uses only the standard library plus cryptography, so it should run unchanged on Linux and macOS, but the outputs below were produced on Windows.
• Packages: cryptography 50.0.2 and pytest 9.2.0.
• Basic Python: functions, classes, running a script. No cloud account, Docker or database server is needed; SQLite ships with Python.

Create a folder, a virtual environment and install the two packages. On Windows (PowerShell):

mkdir auditlog_lab
cd auditlog_lab
python -m venv .venv
.venv\Scripts\Activate.ps1
python -m pip install cryptography==50.0.2 pytest==9.2.0

On Linux or macOS the activation line is source .venv/bin/activate. Every file below goes in this one folder, and every command runs from it.

Step 1: See why an ordinary log proves nothing

Start with the problem. Every demo in this tutorial logs the same six actions of an imaginary agent called agent-7. Look closely at entries 3 and 4: the agent reads a secrets file and then uploads something to an outside server. Save this as events.py. It also holds a fake clock, so timestamps and therefore hashes are the same on every run, and a helper that prints one short line per entry.

# events.py
"""The agent actions every step of the tutorial writes to a log, and a fake clock so the output is repeatable."""

EVENTS = [
    {"actor": "agent-7", "action": "read_file", "target": "/srv/app/config.yaml"},
    {"actor": "agent-7", "action": "run_command", "target": "pytest -q"},
    {"actor": "agent-7", "action": "read_file", "target": "/srv/app/.env"},
    {"actor": "agent-7", "action": "http_post", "target": "https://files.example.net/upload (48211 bytes)"},
    {"actor": "agent-7", "action": "delete_file", "target": "/tmp/work/out.tar"},
    {"actor": "agent-7", "action": "run_command", "target": "git status"},
]


def stamp(n):
    """A repeatable timestamp for entry number n (1-based), one minute apart."""
    return f"2026-10-30T09:{n:02d}:00Z"


def show(record):
    """One short line per entry, for the tutorial output."""
    return f"{record['seq']:>2}  {record['action']:<18}{record['target']}"

Now the log most programs write: one JSON object per line, appended as events happen. Save this as plainlog.py.

# plainlog.py
"""The kind of log most programs write: one JSON object per line, with nothing to check it against."""
import json

from events import EVENTS, stamp


def write_log(path, events=EVENTS):
    with open(path, "w", encoding="utf-8", newline="\n") as handle:
        for n, event in enumerate(events, start=1):
            handle.write(json.dumps({"seq": n, "ts": stamp(n), **event}, sort_keys=True) + "\n")


def read_log(path):
    with open(path, encoding="utf-8") as handle:
        return [json.loads(text) for text in handle if text.strip()]

The demo writes the log, then plays an attacker who can write to the file. The attacker deletes entries 3 and 4 and renumbers the rest so that no gap gives the game away. Save this as demo1_plain.py and run it.

# demo1_plain.py
"""Step 1: edit a plain log so the agent's secret read and upload disappear, then ask whether anything can notice."""
import json

from events import show
from plainlog import read_log, write_log

write_log("plain.jsonl")
print("The log as the agent wrote it:")
for record in read_log("plain.jsonl"):
    print(" ", show(record))

# The attacker can write to the file. Drop entries 3 and 4, then renumber the rest so there is no gap.
kept = [record for record in read_log("plain.jsonl") if record["seq"] not in (3, 4)]
for number, record in enumerate(kept, start=1):
    record["seq"] = number
with open("plain.jsonl", "w", encoding="utf-8", newline="\n") as handle:
    for record in kept:
        handle.write(json.dumps(record, sort_keys=True) + "\n")

print("\nThe log after the edit:")
for record in read_log("plain.jsonl"):
    print(" ", show(record))
print(f"\n{len(read_log('plain.jsonl'))} entries, numbered 1 to {len(kept)}, and nothing to check them against.")

python demo1_plain.py

The log as the agent wrote it:
   1  read_file         /srv/app/config.yaml
   2  run_command       pytest -q
   3  read_file         /srv/app/.env
   4  http_post         https://files.example.net/upload (48211 bytes)
   5  delete_file       /tmp/work/out.tar
   6  run_command       git status

The log after the edit:
   1  read_file         /srv/app/config.yaml
   2  run_command       pytest -q
   3  delete_file       /tmp/work/out.tar
   4  run_command       git status

4 entries, numbered 1 to 4, and nothing to check them against.

The edited log is perfectly consistent: four entries, numbered 1 to 4, each a valid JSON object. Nothing in the file can tell you that two entries ever existed, and that is the problem every later step solves.

Step 2: Chain the entries with hashes

The first defence is a hash chain. Each entry stores a field called prev, the hash of the entry before it, and then gets a hash of its own that covers every other field, prev included. The first entry points at a made-up all-zero hash called the genesis value. Edit any entry and its hash no longer matches its contents. Delete or reorder entries and the prev of the next one no longer matches the hash of the one before it.

2a. Decide exactly which bytes get hashed

Hashing is only as strong as your agreement about what was hashed. If the writer hashes one rendering of a record and a checker hashes another, you get false alarms, or worse, you leave room for two different files that verify identically. The fix is a canonical form: one record has exactly one byte string. Here that means sorted keys, no spaces, ASCII only (anything else becomes a \uXXXX escape), and no NaN values. This is the same discipline Trail of Bits’ SequenceHash asks of anyone who hashes several inputs: do not leave the boundaries between fields ambiguous. JSON syntax gives you the boundaries for free, as long as the form is canonical.

Create chainlog.py with the first part. Notice three details. entry_hash leaves out the hash and mac fields (a hash cannot cover itself, and Step 4 adds the MAC afterwards) but includes prev, which is what links the chain. build_record refuses events that use any of the reserved field names; otherwise an event with a field called mac or hash would be silently ignored by the hash. And the timestamp is passed in rather than read from a clock, so your demos are repeatable.

# chainlog.py
"""Step 2: a hash chain. Every entry carries the hash of the entry before it, so changing one entry breaks every later link."""
import hashlib
import json
import os

GENESIS = "0" * 64
RESERVED = {"seq", "ts", "prev", "hash", "mac"}


class LogFormatError(ValueError):
    """The file holds something that is not exactly what this program would have written."""


def canonical(record):
    """The exact bytes that get hashed and stored: sorted keys, no spaces, ASCII only, no NaN.
    One record has exactly one byte string, so there is nothing to argue about later."""
    text = json.dumps(record, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False)
    return text.encode("ascii")


def entry_hash(record):
    """SHA-256 over the record without its own hash and mac. The prev field IS included: that is the link."""
    body = {key: value for key, value in record.items() if key not in ("hash", "mac")}
    return hashlib.sha256(canonical(body)).hexdigest()


def build_record(prev, seq, ts, event):
    clash = RESERVED & set(event)
    if clash:
        raise ValueError(f"event uses reserved field names: {sorted(clash)}")
    record = {"seq": seq, "ts": ts, **event, "prev": prev}
    record["hash"] = entry_hash(record)
    return record

2b. Write and read the file strictly

The second part is the file format. write_line appends one canonical line, and can call os.fsync afterwards, because Python’s own documentation says that to get data onto the disk you first flush the file object and then call os.fsync(f.fileno()), “to ensure that all internal buffers associated with f are written to disk”. parse_line is the strict reader, and its one rule matters more than it looks: a line is accepted only if it is byte for byte what the writer would have produced.

Why so strict? RFC 8259, the JSON standard, says: “The names within an object SHOULD be unique.” It adds: “When the names within an object are not unique, the behavior of software that receives such an object is unpredictable.” It goes on to say that many implementations report only the last pair, others report an error, and some report every pair. A checker that works on parsed objects therefore cannot be sure that what it verified is what another tool will show. The seventh edit in Step 2d makes the point: it adds a second field to a line, and a plain json.loads still returns the original record. Comparing the stored text with the canonical form of the parsed record rejects duplicate keys, extra spaces, alternative escapes and NaN in one rule, because none of them can appear in canonical output.

# chainlog.py (continued)
def line(record):
    return canonical(record).decode("ascii") + "\n"


def parse_line(text):
    """Read one stored line strictly: the text must be the canonical form, byte for byte.
    One rule covers duplicate keys, extra spaces, other escapes and NaN, because none of them can appear in canonical output."""
    record = json.loads(text)
    if not isinstance(record, dict) or text != canonical(record).decode("ascii"):
        raise LogFormatError("record is not in canonical form")
    return record


def read_records(path):
    if not os.path.exists(path):
        return []
    with open(path, encoding="ascii", newline="") as handle:
        return [parse_line(text) for text in handle.read().split("\n") if text]


def write_line(path, record, durable=False):
    with open(path, "a", encoding="ascii", newline="\n") as handle:
        handle.write(line(record))
        if durable:
            handle.flush()
            os.fsync(handle.fileno())

2c. The writer and the verifier

The writer remembers the sequence number and hash of the last entry (it reads them from the file when it starts, so you can restart it) and appends. The verifier walks the entries once and checks three things in order: the sequence number is the expected one, prev equals the previous entry’s hash, and the stored hash equals a fresh hash of the entry. It reports the first problem it finds. Add this to the same file.

# chainlog.py (continued)
class ChainWriter:
    """Appends entries to a JSON Lines file. One writer process only; Step 7 shows what two of them do."""

    def __init__(self, path, durable=False):
        self.path, self.durable = path, durable
        records = read_records(path)
        self.seq = records[-1]["seq"] if records else 0
        self.head = records[-1]["hash"] if records else GENESIS

    def append(self, event, ts):
        record = build_record(self.head, self.seq + 1, ts, event)
        write_line(self.path, record, self.durable)
        self.seq, self.head = record["seq"], record["hash"]
        return record


def verify_chain(records):
    """Return (True, summary) or (False, why) for the FIRST problem found."""
    prev = GENESIS
    for index, record in enumerate(records, start=1):
        if record.get("seq") != index:
            return False, f"seq {record.get('seq')}: expected seq {index}"
        if record.get("prev") != prev:
            return False, f"seq {index}: prev does not match the entry before it"
        if record.get("hash") != entry_hash(record):
            return False, f"seq {index}: hash mismatch (the entry was changed)"
        prev = record["hash"]
    return True, f"{len(records)} entries, head {prev[:12]}"

The demos need a few small helpers: a function that writes the six sample events through a chain writer, and two that read and write a file as raw lines, which is how an attacker would touch it. Save this as labkit.py.

# labkit.py
"""Helpers shared by the demos and the tests."""
from chainlog import ChainWriter
from events import EVENTS, stamp


class Clock:
    """Hands out stamp(1), stamp(2), ... so a log written through a hook has repeatable timestamps."""

    def __init__(self):
        self.n = 0

    def __call__(self):
        self.n += 1
        return stamp(self.n)


def fresh_chain(path):
    open(path, "w").close()
    writer = ChainWriter(path)
    for n, event in enumerate(EVENTS, start=1):
        writer.append(event, stamp(n))
    return writer


def read_lines(path):
    with open(path, encoding="ascii", newline="") as handle:
        return handle.read().split("\n")[:-1]


def write_lines(path, lines):
    with open(path, "w", encoding="ascii", newline="\n") as handle:
        for text in lines:
            handle.write(text + "\n")

2d. Attack the chain

This demo writes a good log and then applies six edits to the file, one at a time, to a fresh copy. After each edit it asks verify_chain for a verdict. The seventh edit is the sneaky one: it adds a second "action" field to line 3. The demo hands that line to a lenient reader (plain json.loads) and to the strict one. Save it as demo2_attacks.py and run it.

# demo2_attacks.py
"""Step 2: write a chained log, try seven edits on the file, and let the verifier judge each one."""
import json

from chainlog import LogFormatError, parse_line, read_records, verify_chain
from labkit import fresh_chain, read_lines, write_lines


def edit_entry_3(lines):
    lines[2] = lines[2].replace("/srv/app/.env", "/srv/app/notes.txt")
    return lines


def delete_entry_4(lines):
    del lines[3]
    return lines


def delete_4_and_renumber(lines):
    records = [json.loads(text) for text in lines]
    del records[3]
    for number, record in enumerate(records, start=1):
        record["seq"] = number
    return [json.dumps(record, sort_keys=True, separators=(",", ":")) for record in records]


def swap_2_and_3(lines):
    lines[1], lines[2] = lines[2], lines[1]
    return lines


def insert_forged_entry(lines):
    forged = json.loads(lines[3])
    forged.update(seq=5, action="read_file", target="/srv/app/notes.txt")
    return lines[:4] + [json.dumps(forged, sort_keys=True, separators=(",", ":"))] + lines[4:]


def cut_off_last_two(lines):
    return lines[:4]


ATTACKS = [
    ("edit entry 3 in place", edit_entry_3),
    ("delete entry 4", delete_entry_4),
    ("delete entry 4 and renumber", delete_4_and_renumber),
    ("swap entries 2 and 3", swap_2_and_3),
    ("insert a forged entry after 4", insert_forged_entry),
    ("cut off the last two entries", cut_off_last_two),
]

fresh_chain("chain.jsonl")
print("untouched log:".ljust(34), verify_chain(read_records("chain.jsonl")))
original = read_lines("chain.jsonl")
for name, attack in ATTACKS:
    write_lines("attacked.jsonl", attack(list(original)))
    ok, message = verify_chain(read_records("attacked.jsonl"))
    print(f"{name:<34}", "OK" if ok else "FAIL", "-", message)

# The seventh edit: a duplicate key on line 3. Most parsers keep the last value, so the parsed record looks unchanged.
sneaky = original[2].replace('{"action":"read_file"', '{"action":"read_file","action":"read_file"', 1)
print("\nduplicate key, lenient reader:".ljust(35), verify_chain([json.loads(text) for text in original[:2] + [sneaky]]))
try:
    parse_line(sneaky)
except LogFormatError as error:
    print("duplicate key, strict reader:".ljust(34), "FAIL -", error)

python demo2_attacks.py

untouched log:                     (True, '6 entries, head b4620e52dceb')
edit entry 3 in place              FAIL - seq 3: hash mismatch (the entry was changed)
delete entry 4                     FAIL - seq 5: expected seq 4
delete entry 4 and renumber        FAIL - seq 4: prev does not match the entry before it
swap entries 2 and 3               FAIL - seq 3: expected seq 2
insert a forged entry after 4      FAIL - seq 5: prev does not match the entry before it
cut off the last two entries       OK - 4 entries, head 7ed07befce2a

duplicate key, lenient reader:     (True, '3 entries, head dea14739046f')
duplicate key, strict reader:      FAIL - record is not in canonical form

Read the table carefully, because it teaches three things.

1. Edits, deletions, swaps and inserts are caught, and the message points at the first broken entry. Deleting entry 4 is reported at seq 5, because that is the first entry that no longer fits.
2. Cutting off the end is not caught. The last two entries are gone and the verdict is OK, because a chain only checks that what is present is consistent. Nothing in the file says how long it should be. Hold on to that row; Step 6 exists because of it.
3. The lenient reader missed the duplicate key. It parsed the doctored line 3 into the same record as the original and said OK. The strict reader refused the line. A verifier should check the bytes it stored, not the object it happened to parse.

Step 3: Watch an attacker rebuild the whole chain

The chain caught every clumsy edit, so is it enough? Ask what the attacker needs to rewrite history convincingly. Nothing secret goes into a hash: the algorithm is public, and your own ChainWriter is sitting right there. An attacker who can write the file can drop entries 3 and 4 and then replay the surviving events through the same code, with their original timestamps, to produce a brand-new chain with correct hashes from top to bottom. Save this as demo3_rechain.py and run it.

# demo3_rechain.py
"""Step 3: an attacker who can run the same code rebuilds the whole chain, and the verifier is satisfied."""
import os

from chainlog import ChainWriter, read_records, verify_chain
from events import show
from labkit import fresh_chain

fresh_chain("chain.jsonl")
original = read_records("chain.jsonl")
print("original: ", verify_chain(original))

# Drop entries 3 and 4, then write the survivors again through the public algorithm with their old timestamps.
if os.path.exists("rebuilt.jsonl"):
    os.remove("rebuilt.jsonl")
writer = ChainWriter("rebuilt.jsonl")
for record in original:
    if record["seq"] not in (3, 4):
        writer.append({key: record[key] for key in ("actor", "action", "target")}, record["ts"])

rebuilt = read_records("rebuilt.jsonl")
print("rebuilt:  ", verify_chain(rebuilt))
for record in rebuilt:
    print(" ", show(record))

python demo3_rechain.py

original:  (True, '6 entries, head b4620e52dceb')
rebuilt:   (True, '4 entries, head cf2dada5b2f3')
   1  read_file         /srv/app/config.yaml
   2  run_command       pytest -q
   3  delete_file       /tmp/work/out.tar
   4  run_command       git status

The rebuilt log verifies. The secret read and the upload are gone, the head hash is different, and the verifier is satisfied. So an unkeyed hash chain protects against accidents such as a truncated write, a bad disk sector or a clumsy edit. It does not protect against an attacker who can run code. To stop that, the attacker must be missing something they cannot compute.

Step 4: Add a secret key with HMAC

An HMAC is a hash that is mixed with a secret key. Give every entry a mac field, the HMAC of the entry’s hash, and an attacker who does not hold the key cannot produce valid MACs for entries they rewrite. The verifier, which holds the same key, recomputes every MAC and compares it using hmac.compare_digest, which takes the same time whether the first character or the last one differs, so the comparison does not leak how close a guess was (the webhook signature tutorial measures that effect).

Two small habits in the code below are worth copying. Both functions mix a fixed label (auditlog/mac/v1 and auditlog/evolve/v1) into the hashed bytes, so a value produced for one purpose can never be mistaken for a value produced for another. And SealedWriter keeps its key in a bytearray, which can be overwritten in place, rather than in an immutable bytes object. The evolve_keys switch is already in this file so you type it only once; in this step it stays False, which means one fixed key for every entry. Save this as sealedlog.py.

# sealedlog.py
"""Steps 4 and 5: a keyed MAC on every entry, and a key that changes after every entry."""
import hashlib
import hmac

from chainlog import GENESIS, build_record, read_records, verify_chain, write_line


def evolve(key):
    """The next key. SHA-256 only runs forwards, so a stolen key cannot be turned back into an earlier one."""
    return hashlib.sha256(b"auditlog/evolve/v1" + key).digest()


def tag(key, record_hash):
    """HMAC-SHA256 of an entry's hash. Only someone holding the key can produce it."""
    return hmac.new(key, b"auditlog/mac/v1" + bytes.fromhex(record_hash), hashlib.sha256).hexdigest()


class SealedWriter:
    def __init__(self, path, key, evolve_keys=True, durable=False):
        self.path, self.evolve_keys, self.durable = path, evolve_keys, durable
        self._key = bytearray(key)  # a bytearray, so the old key can be overwritten in place
        records = read_records(path)
        self.seq = records[-1]["seq"] if records else 0
        self.head = records[-1]["hash"] if records else GENESIS

    def current_key(self):
        return bytes(self._key)

    def append(self, event, ts):
        record = build_record(self.head, self.seq + 1, ts, event)
        record["mac"] = tag(bytes(self._key), record["hash"])
        write_line(self.path, record, self.durable)
        self.seq, self.head = record["seq"], record["hash"]
        if self.evolve_keys:
            self._key[:] = evolve(bytes(self._key))  # the key that sealed this entry is gone after this line
        return record


def verify_sealed(records, key, evolve_keys=True):
    """Check the chain, then check that every entry was sealed with the key it should have been sealed with."""
    ok, message = verify_chain(records)
    if not ok:
        return ok, message
    for index, record in enumerate(records, start=1):
        expected = tag(key, record["hash"])
        if not hmac.compare_digest(str(record.get("mac", "")).encode(), expected.encode()):
            return False, f"seq {index}: bad mac (not sealed by the key holder)"
        if evolve_keys:
            key = evolve(key)
    return True, f"{message}, macs ok"

The attacker’s toolbox goes in its own file, because the next three demos all use it. forge models what someone holding a key can do: keep the first few entries and append new ones that carry valid hashes and MACs. key_at(n) derives the key that seals entry n the way the verifier does. KEY0 is derived from a fixed string only so that your output matches this page; a real key comes from secrets.token_bytes(32) and lives in a secret manager or a hardware security module. SIGNER is a fixed Ed25519 signing key that Step 6 uses. Save this as attacker.py.

# attacker.py
"""The fixed keys used by the sealed-log demos, and the attacker's toolbox."""
import hashlib

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from chainlog import GENESIS, build_record
from events import EVENTS, stamp
from sealedlog import SealedWriter, evolve, tag

# Fixed so the output is repeatable. A real key comes from secrets.token_bytes(32) and lives in a secret manager or an HSM.
KEY0 = hashlib.sha256(b"tutorial-demo-key-not-for-real-use").digest()
SIGNER = Ed25519PrivateKey.from_private_bytes(hashlib.sha256(b"tutorial-signing-seed").digest())


def fresh_sealed(path, evolve_keys=True):
    open(path, "w").close()
    writer = SealedWriter(path, KEY0, evolve_keys=evolve_keys)
    for n, event in enumerate(EVENTS, start=1):
        writer.append(event, stamp(n))
    return writer


def key_at(n):
    """The key that seals entry n (1-based), derived from KEY0 the way the verifier derives it."""
    key = KEY0
    for _ in range(n - 1):
        key = evolve(key)
    return key


def forge(prefix, events, key, evolve_keys=True):
    """What an attacker who holds `key` can do: keep `prefix` and add new, validly sealed entries after it."""
    records = list(prefix)
    for event, ts in events:
        prev = records[-1]["hash"] if records else GENESIS
        record = build_record(prev, len(records) + 1, ts, event)
        record["mac"] = tag(key, record["hash"])
        records.append(record)
        if evolve_keys:
            key = evolve(key)
    return records

The demo writes six sealed entries with one static key. Then it plays two attackers who both keep entries 1 and 2, drop 3 and 4, and re-create the last two events as the new entries 3 and 4. One has no key. The other has stolen it. Save this as demo4_hmac.py and run it.

# demo4_hmac.py
"""Step 4: key the chain with an HMAC. A rebuild fails without the key and succeeds with it."""
from chainlog import read_records
from events import EVENTS, stamp
from attacker import KEY0, fresh_sealed, forge
from sealedlog import verify_sealed

fresh_sealed("sealed_static.jsonl", evolve_keys=False)
original = read_records("sealed_static.jsonl")
print("original:                     ", verify_sealed(original, KEY0, evolve_keys=False))

# Keep entries 1 and 2, drop 3 and 4, and re-create entries 5 and 6 as the new 3 and 4.
later = [(EVENTS[4], stamp(5)), (EVENTS[5], stamp(6))]
no_key = forge(original[:2], later, b"\x00" * 32, evolve_keys=False)
print("rebuilt without the key:      ", verify_sealed(no_key, KEY0, evolve_keys=False))

stolen = forge(original[:2], later, KEY0, evolve_keys=False)
print("rebuilt with the stolen key:  ", verify_sealed(stolen, KEY0, evolve_keys=False))

python demo4_hmac.py

original:                      (True, '6 entries, head b4620e52dceb, macs ok')
rebuilt without the key:       (False, 'seq 3: bad mac (not sealed by the key holder)')
rebuilt with the stolen key:   (True, '4 entries, head cf2dada5b2f3, macs ok')

The attacker without the key fails at seq 3, the first entry they had to change; entries 1 and 2 are untouched and still verify. The attacker with the key succeeds, and the verdict is the same OK as for an honest log. That second line is the uncomfortable one. The logging process has to hold the key to seal entries, so an attacker who takes over the logging host can take the key too, and then rewrite everything, past and future.

Step 5: Change the key after every entry

This is the idea behind forward security, and Bruce Schneier and John Kelsey described it for audit logs. Their paper, Secure Audit Logs to Support Computer Forensics, sets out to make “all log entries generated prior to the logging machine’s compromise impossible for the attacker to read, and also impossible to undetectably modify or destroy.” The mechanism is short. In the authors’ words the log’s authentication key is hashed “immediately after a log entry is written” and the new value “overwrites and irretrieveably deletes the previous value”. (The spelling is theirs.) Because a hash only runs forwards, the machine can step the key from entry n to entry n+1 but nobody can step it back.

The same family of ideas ships in real systems. The systemd journal’s Seal= setting controls Forward Secure Sealing, which the journald.conf documentation says “may be used to protect journal files from unnoticed alteration”.

You already typed the code. Look back at sealedlog.py: evolve(key) is the one-way step, the writer overwrites its key in place with self._key[:] = evolve(bytes(self._key)) right after sealing an entry, and verify_sealed repeats key = evolve(key) as it walks the log, starting from the initial key that the verifier keeps offline. The default evolve_keys=True turns all of that on. The verifier is the one party that can regenerate every key, because it holds KEY0.

The demo writes six entries with evolving keys. Then it simulates a memory dump of the logging process after entry 6, which hands the attacker the key for entry 7. It tries three things with that key. Save it as demo5_forward.py and run it.

# demo5_forward.py
"""Step 5: change the key after every entry. A key stolen now cannot rewrite what was sealed before."""
from chainlog import read_records
from events import EVENTS, stamp
from attacker import KEY0, fresh_sealed, forge, key_at
from sealedlog import verify_sealed

writer = fresh_sealed("sealed_evolving.jsonl")
original = read_records("sealed_evolving.jsonl")
print("original:                          ", verify_sealed(original, KEY0))

stolen = writer.current_key()  # what a memory dump taken after entry 6 would contain
print("stolen key is the key for entry 7: ", stolen == key_at(7), "| it is not the initial key:", stolen != KEY0)

later = [(EVENTS[4], stamp(5)), (EVENTS[5], stamp(6))]
print("rewrite entries 3 and 4, stolen key:", verify_sealed(forge(original[:2], later, stolen), KEY0))

extra = [({"actor": "agent-7", "action": "send_email", "target": "all-staff"}, stamp(7))]
print("append entry 7 with the stolen key: ", verify_sealed(forge(original, extra, stolen), KEY0))
print("cut off the last two entries:       ", verify_sealed(original[:4], KEY0))

python demo5_forward.py

original:                           (True, '6 entries, head b4620e52dceb, macs ok')
stolen key is the key for entry 7:  True | it is not the initial key: True
rewrite entries 3 and 4, stolen key: (False, 'seq 3: bad mac (not sealed by the key holder)')
append entry 7 with the stolen key:  (True, '7 entries, head 6fc359952533, macs ok')
cut off the last two entries:        (True, '4 entries, head 7ed07befce2a, macs ok')

Rewriting entries 3 and 4 fails at seq 3: the attacker’s MAC used the key for entry 7, and the verifier expects the key for entry 3, which the stolen key cannot produce. The past is safe. Two things are not safe, and the last two lines say so plainly:

• The future. With the stolen key the attacker can seal as many new entries as they like after entry 6, and the log verifies. Forward security protects what was written before the compromise, not what is written after it.
• The end of the log. Cutting off the last two entries still verifies. The key chain does not know how many entries there should be.

One honest caveat about Python. Overwriting a bytearray erases that buffer, but bytes(self._key) makes temporary copies, the interpreter may leave others behind, and a crash dump or swap file can capture them. Real deployments keep the evolving key in a hardened process or a hardware module. The design is sound; the Python is a demonstration.

Step 6: Sign checkpoints and keep them somewhere else

Both gaps, truncation and the post-compromise future, have the same cure: a reference value stored where the attacker cannot reach it. Schneier and Kelsey name the truncation problem themselves, in their section on abnormal shutdowns: “it will be impossible to distinguish the log of a system that crashed at 12:10 PM and the log of a system that was taken over at 2:00 PM, and then had all records entered since 12:10 PM truncated.” Their own fix is a special message kept in nonvolatile storage and deleted when the next entry is written. A more common one, and the one you will build, is to publish a signed summary of the log’s head from time to time.

Amazon CloudTrail does this at industry scale. Its log file integrity validation uses “SHA-256 for hashing and SHA-256 with RSA for digital signing”, delivers a signed digest file every hour, and states that “Each digest file also contains the digital signature of the previous digest file if one exists.” That is a hash chain of signed checkpoints, kept in a separate folder from the logs so that separate access rules can apply.

A checkpoint here says: log agent-actions had an entry number seq and its hash was head. It is signed with an Ed25519 private key; anyone with the public key can check it, and nobody without the private key can forge one. The signed bytes begin with a fixed label for the same reason as before. Checking a checkpoint against a log means three tests in order: the signature is valid, the log is at least as long as the checkpoint says, and the entry at that position still has the checkpoint’s hash. Save this as anchor.py.

# anchor.py
"""Step 6: signed checkpoints that live somewhere else, so cutting the end off the log can be noticed."""
from cryptography.exceptions import InvalidSignature

from chainlog import canonical

DOMAIN = b"auditlog/checkpoint/v1\n"


def make_checkpoint(private_key, log_id, record):
    """Sign: 'log <log_id> had an entry number <seq> and its hash was <head>'."""
    body = {"log": log_id, "seq": record["seq"], "head": record["hash"]}
    return {**body, "sig": private_key.sign(DOMAIN + canonical(body)).hex()}


def check_checkpoint(public_key, checkpoint, records):
    """Return (True, summary) or (False, why). The signature is checked first, then the log is compared with it."""
    try:
        body = {name: checkpoint[name] for name in ("log", "seq", "head")}
        public_key.verify(bytes.fromhex(checkpoint["sig"]), DOMAIN + canonical(body))
    except (InvalidSignature, ValueError, KeyError, TypeError):
        return False, "checkpoint signature is not valid"
    if len(records) < body["seq"]:
        return False, f"log has {len(records)} entries but a signed checkpoint covers {body['seq']} (the end was cut off)"
    if records[body["seq"] - 1]["hash"] != body["head"]:
        return False, f"entry {body['seq']} no longer matches the checkpoint (history was rewritten)"
    return True, f"checkpoint at seq {body['seq']} still holds"

The demo writes six sealed entries and signs a checkpoint at entry 4, as if the logger did so every four entries. It saves the checkpoint in a folder called witness, which stands in for another machine, a storage bucket with write-once retention or a public transparency log. It then runs five scenarios and prints a chain verdict and an anchor verdict for each. The fourth scenario is the dangerous one: the attacker stole the key for entry 3 before entry 3 was written, so they can seal a whole new tail. Save it as demo6_anchor.py and run it.

# demo6_anchor.py
"""Step 6: sign a checkpoint every 4 entries and keep it somewhere the log host cannot reach."""
import json
import os

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from anchor import check_checkpoint, make_checkpoint
from chainlog import read_records
from events import EVENTS, stamp
from attacker import KEY0, SIGNER, fresh_sealed, forge, key_at
from sealedlog import verify_sealed

fresh_sealed("anchored.jsonl")
records = read_records("anchored.jsonl")
os.makedirs("witness", exist_ok=True)  # stands in for another machine, a bucket with object lock, or a transparency log
with open("witness/checkpoint_4.json", "w", encoding="ascii") as handle:
    json.dump(make_checkpoint(SIGNER, "agent-actions", records[3]), handle)
with open("witness/checkpoint_4.json", encoding="ascii") as handle:
    checkpoint = json.load(handle)
public_key = SIGNER.public_key()


def judge(label, candidate):
    chain = verify_sealed(candidate, KEY0)
    anchor = check_checkpoint(public_key, checkpoint, candidate)
    print(f"{label:<38} chain: {'OK  ' if chain[0] else 'FAIL'}  anchor: {'OK  ' if anchor[0] else 'FAIL'} {anchor[1]}")


judge("untouched", records)
judge("cut back to 5 entries", records[:5])
judge("cut back to 3 entries", records[:3])

# Compromise before entry 3 was sealed: the attacker holds key_at(3), so they can seal a whole new tail.
rewritten = forge(records[:2], [(EVENTS[0], stamp(3)), (EVENTS[1], stamp(4)), (EVENTS[5], stamp(5)), (EVENTS[5], stamp(6))], key_at(3))
judge("tail rewritten with the stolen key", rewritten)

# A checkpoint the attacker signs with their own key.
fake = make_checkpoint(Ed25519PrivateKey.generate(), "agent-actions", rewritten[3])
print("checkpoint signed by the attacker:     ", check_checkpoint(public_key, fake, rewritten))

python demo6_anchor.py

untouched                              chain: OK    anchor: OK   checkpoint at seq 4 still holds
cut back to 5 entries                  chain: OK    anchor: OK   checkpoint at seq 4 still holds
cut back to 3 entries                  chain: OK    anchor: FAIL log has 3 entries but a signed checkpoint covers 4 (the end was cut off)
tail rewritten with the stolen key     chain: OK    anchor: FAIL entry 4 no longer matches the checkpoint (history was rewritten)
checkpoint signed by the attacker:      (False, 'checkpoint signature is not valid')

Read it row by row:

• Cut back to 5 entries: both checks pass. The checkpoint covers entry 4, so entries 5 and 6 are not protected. This is the exposure window: whatever was written after the last checkpoint you have stored. Anchoring every hour leaves an hour; anchoring after every entry leaves nothing, at the cost of a network call per entry.
• Cut back to 3 entries: the anchor fails with a message that the end was cut off, while the chain alone still says OK.
• Tail rewritten with a stolen key: chain OK, anchor FAIL. The attacker’s forgery is perfectly valid as a chain, but entry 4 no longer has the hash that the checkpoint recorded before the rewrite. This is the case that forward security could not handle, and it is why the checkpoint must be exported before the attacker arrives.
• A checkpoint signed by the attacker is rejected, because it does not verify with the real public key.

The design depends on one assumption: the attacker cannot delete or replace the stored checkpoints. If the witness folder sits on the same host as the log, an attacker simply removes it and every row above becomes “OK”. Keep checkpoints on another machine, in storage with write-once retention, or in a transparency log, and make your verifier remember the newest checkpoint it has ever accepted, so that rolling both the log and its checkpoints back to an older backup is noticed.

Step 7: Choose where the log lives, in SQLite

Two practical problems remain, and a database file solves one of them and exposes the other.

Problem one: two writers. A chain is a single line of entries. If two processes each remember the last hash in memory and both append, they both build entry 7 on top of entry 6, and the chain forks. Worse, an agent action can end up with no entry at all. The cure is to make the read the head, then append step happen under a lock. SQLite has what you need: per its documentation, “IMMEDIATE causes the database connection to start a new write immediately, without waiting for a write statement,” and “The BEGIN IMMEDIATE might fail with SQLITE_BUSY if another write transaction is already active on another database connection.” With BEGIN IMMEDIATE, a second writer waits for the first to commit, then reads the new head.

Problem two: protection inside the database. Triggers can make the table append-only, so an UPDATE or DELETE from the application, or from an SQL injection bug in it, is refused. That is worth having, and it is the same idea as database permissions in our multi-tenant audit logging tutorial. But a trigger is only a rule inside a file. Whoever can open the file with a plain SQLite connection can drop the trigger and edit the row, so the chain has to remain the thing that notices.

Save this as sqlitelog.py. append takes a locked flag so you can compare the two behaviours; append_many is the worker the demo hands to several processes, and it lives in a module because Windows starts worker processes by importing it. The primary key on seq matters: if two writers ever pick the same number, the database refuses the second one instead of storing a forked chain.

# sqlitelog.py
"""Step 7: keep the log in SQLite so that several writer processes take turns instead of forking the chain."""
import json
import sqlite3

from chainlog import GENESIS, build_record, canonical, parse_line

SCHEMA = """
CREATE TABLE IF NOT EXISTS log (seq INTEGER PRIMARY KEY, body TEXT NOT NULL);
CREATE TRIGGER IF NOT EXISTS log_no_update BEFORE UPDATE ON log BEGIN SELECT RAISE(ABORT, 'log is append-only'); END;
CREATE TRIGGER IF NOT EXISTS log_no_delete BEFORE DELETE ON log BEGIN SELECT RAISE(ABORT, 'log is append-only'); END;
"""


def connect(path):
    db = sqlite3.connect(path, timeout=30, isolation_level=None)  # autocommit: we send BEGIN ourselves
    db.executescript(SCHEMA)
    return db


def head(db):
    row = db.execute("SELECT body FROM log ORDER BY seq DESC LIMIT 1").fetchone()
    if row is None:
        return 0, GENESIS
    last = json.loads(row[0])
    return last["seq"], last["hash"]


def append(db, event, ts, locked=True):
    """locked=True takes SQLite's write lock BEFORE reading the head, so two writers can never read the same head."""
    if locked:
        db.execute("BEGIN IMMEDIATE")
    try:
        seq, prev = head(db)
        record = build_record(prev, seq + 1, ts, event)
        db.execute("INSERT INTO log (seq, body) VALUES (?, ?)", (record["seq"], canonical(record).decode("ascii")))
        if locked:
            db.execute("COMMIT")
        return record
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise


def records(db):
    return [parse_line(body) for (body,) in db.execute("SELECT body FROM log ORDER BY seq")]


def append_many(path, who, count, locked):
    """Worker for the multi-process demo and test: returns how many appends failed."""
    db = connect(path)
    failed = 0
    for i in range(count):
        try:
            append(db, {"actor": f"agent-{who}", "action": "tick", "target": str(i)}, f"2026-10-30T10:{who:02d}:{i:02d}Z", locked=locked)
        except sqlite3.Error:
            failed += 1
    db.close()
    return failed

The demo does three things. It appends six entries and tries an UPDATE and a DELETE through the application’s own connection. It then opens the file directly, drops the update trigger and edits entry 3. Finally it starts four processes that each append 25 entries, once with the lock and once without. Save it as demo7_sqlite.py and run it.

# demo7_sqlite.py
"""Step 7: triggers stop the application, a file-level attacker drops them, and only the chain notices."""
import multiprocessing
import os
import sqlite3

import sqlitelog
from chainlog import verify_chain
from events import EVENTS, stamp


def main():
    if os.path.exists("agent.db"):
        os.remove("agent.db")
    db = sqlitelog.connect("agent.db")
    for n, event in enumerate(EVENTS, start=1):
        sqlitelog.append(db, event, stamp(n))
    print("after 6 appends:           ", verify_chain(sqlitelog.records(db)))
    for statement in ("UPDATE log SET body = 'x' WHERE seq = 3", "DELETE FROM log WHERE seq = 3"):
        try:
            db.execute(statement)
        except sqlite3.Error as error:
            print(f"{statement[:6]} through the app:".ljust(27), error)

    # Someone with the database file opens it directly, drops the triggers and edits entry 3.
    raw = sqlite3.connect("agent.db", isolation_level=None)
    raw.execute("DROP TRIGGER log_no_update")
    raw.execute("UPDATE log SET body = replace(body, '.env', 'notes.txt') WHERE seq = 3")
    raw.close()
    print("after the file-level edit: ", verify_chain(sqlitelog.records(db)))
    db.close()

    for locked in (True, False):
        if os.path.exists("race.db"):
            os.remove("race.db")
        sqlitelog.connect("race.db").close()
        with multiprocessing.Pool(4) as pool:
            failed = sum(pool.starmap(sqlitelog.append_many, [("race.db", who, 25, locked) for who in range(1, 5)]))
        check = sqlitelog.connect("race.db")
        ok, _ = verify_chain(sqlitelog.records(check))
        stored = len(sqlitelog.records(check))
        check.close()
        note = "" if locked else " (varies per run)"
        print(f"4 processes x 25, locked={locked!s:<5}: stored {stored}, failed {failed}, chain ok {ok}{note}")


if __name__ == "__main__":
    main()

python demo7_sqlite.py

after 6 appends:            (True, '6 entries, head b4620e52dceb')
UPDATE through the app:     log is append-only
DELETE through the app:     log is append-only
after the file-level edit:  (False, 'seq 3: hash mismatch (the entry was changed)')
4 processes x 25, locked=True : stored 100, failed 0, chain ok True
4 processes x 25, locked=False: stored 97, failed 3, chain ok True (varies per run)

The triggers refuse both statements. The file-level edit goes through, and the chain notices: entry 3 no longer matches its hash. The race is the last two lines. With the lock, all 100 entries are stored and none failed. Without it, some appends were rejected because two processes had read the same head and chosen the same sequence number. The count changes from run to run, which is why the line says so; on the run shown, 3 of the 100 actions never made it into the log. When I tallied the errors over three more runs, all eight were UNIQUE constraint failed: log.seq, which is the primary key doing its job. For an audit log that is the worse failure, because nothing in the stored data shows what is missing.

One design note before moving on. This store uses the unkeyed chain, because key evolution needs a single owner of the key and the sequence. In a real deployment you would run one small logging service that owns the evolving key and the SQLite file, and have every agent process send it events over a local socket or pipe. The service is the single writer that the sealed log in Steps 4 to 6 assumes.

Step 8: Log every agent action before it happens

Now connect the log to an agent. The rule that matters here is about ordering: write the entry before the action, not after it. If you log after the tool returns, a crash, a kill or a power cut in the middle of the action leaves a real action with no record. If you log the call first and the result afterwards, the worst case is a call with no result, and that is exactly the entry you want to find.

audited is a decorator for an agent’s tools. It writes NAME.call before running the tool, then NAME.ok or NAME.error afterwards, and it gives each call an id so the two entries can be paired. Save this as agent_hook.py.

# agent_hook.py
"""Step 8: wrap an agent's tools so every call is logged before it runs and again when it ends."""
import functools
import itertools


def audited(writer, clock, actor="agent-7"):
    call_ids = itertools.count(1)

    def decorate(tool):
        @functools.wraps(tool)
        def wrapper(*args, **kwargs):
            note = {"actor": actor, "call_id": next(call_ids)}
            writer.append({**note, "action": f"{tool.__name__}.call", "target": repr(args)[:60]}, clock())
            try:
                result = tool(*args, **kwargs)
            except BaseException as error:
                writer.append({**note, "action": f"{tool.__name__}.error", "target": type(error).__name__}, clock())
                raise
            writer.append({**note, "action": f"{tool.__name__}.ok", "target": ""}, clock())
            return result

        return wrapper

    return decorate

To see the ordering matter, you need a process that dies mid-call. os._exit(1) ends a Python process immediately, without cleanup, which is how a SIGKILL looks from the log’s side. A power cut can be worse, because data that never reached the disk can vanish as well. Save this as crash_child.py.

# crash_child.py
"""Step 8: start a tool call, then die the way a killed process does, without running any cleanup."""
import os
import sys

from agent_hook import audited
from attacker import KEY0
from labkit import Clock
from sealedlog import SealedWriter

writer = SealedWriter(sys.argv[1], KEY0)


audit = audited(writer, Clock())


@audit
def read_file(path):
    return "contents of " + path


@audit
def delete_file(path):
    os._exit(1)  # stands in for a SIGKILL or a power cut in the middle of the action


read_file("/srv/app/config.yaml")
delete_file("/srv/app/data.db")

The demo logs two tool calls in one process, a file read that works and an upload that raises ConnectionError, and prints the entries. Then it runs the crashing child and reads what that left behind. Save it as demo8_agent.py and run it.

# demo8_agent.py
"""Step 8: log an agent's tool calls through a decorator, then kill a process in the middle of a call."""
import os
import subprocess
import sys

from agent_hook import audited
from chainlog import read_records
from events import show
from attacker import KEY0
from labkit import Clock
from sealedlog import SealedWriter, verify_sealed

if os.path.exists("agent_calls.jsonl"):
    os.remove("agent_calls.jsonl")
writer = SealedWriter("agent_calls.jsonl", KEY0)


audit = audited(writer, Clock())


@audit
def read_file(path):
    return "contents of " + path


@audit
def http_post(url, body):
    raise ConnectionError("upload refused")


read_file("/srv/app/config.yaml")
try:
    http_post("https://files.example.net/upload", b"...")
except ConnectionError:
    pass
for record in read_records("agent_calls.jsonl"):
    print(" ", show(record), f"[call {record['call_id']}]")
print("verify:", verify_sealed(read_records("agent_calls.jsonl"), KEY0))

if os.path.exists("agent_crash.jsonl"):
    os.remove("agent_crash.jsonl")
child = subprocess.run([sys.executable, "crash_child.py", "agent_crash.jsonl"])
records = read_records("agent_crash.jsonl")
print(f"\nchild exit code {child.returncode}; the log holds {len(records)} entries, the last one is:", show(records[-1]))
print("verify:", verify_sealed(records, KEY0))

python demo8_agent.py

   1  read_file.call    ('/srv/app/config.yaml',) [call 1]
   2  read_file.ok       [call 1]
   3  http_post.call    ('https://files.example.net/upload', b'...') [call 2]
   4  http_post.error   ConnectionError [call 2]
verify: (True, '4 entries, head 92e32f78cdee, macs ok')

child exit code 1; the log holds 3 entries, the last one is:  3  delete_file.call  ('/srv/app/data.db',)
verify: (True, '3 entries, head 9460eee99244, macs ok')

The failed upload is on the record, with its error type, and so is the crash: the last entry of the child’s log is delete_file.call with no ok or error after it, and the log still verifies. Reading it, you know exactly which action was in flight when the process died.

A warning about what goes into the target field. A log that can never be edited can never be cleaned up either, so anything you write stays there for good, including file contents, tokens and personal data that you might later be required to delete. Log references instead: a file path, a record id, or a salted hash of a sensitive value. The hook above truncates the arguments to 60 characters only to keep the output short; decide what an entry may contain deliberately.

Step 9: Prove the tests would notice a broken verifier

A verifier that silently stops checking something is worse than none, because you keep trusting it. So pin every claim in this tutorial, including the claims about what the design cannot do, with a test. The file below has 54 test cases. Some assert what the design catches: editing any field, deleting any entry but the last, a stolen evolving key failing to rewrite earlier entries, a checkpoint catching a rewritten tail. Others assert the limits on purpose, such as test_cutting_off_the_end_is_invisible_to_the_chain, so that nobody quietly starts relying on a guarantee the code does not give. Two tests deserve a note. One builds an entry whose own hash is right but whose link is wrong, and another builds a last entry whose hash and link are right but whose number is wrong; each check in verify_chain needs its own test, or you cannot tell whether it is doing anything. Save this as test_auditlog.py.

# test_auditlog.py
"""Every test pins one claim the tutorial makes, including the claims about what the design cannot do."""
import json
import multiprocessing
import sqlite3

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

import sqlitelog
from agent_hook import audited
from anchor import check_checkpoint, make_checkpoint
from chainlog import GENESIS, RESERVED, ChainWriter, LogFormatError, build_record, line, parse_line, read_records, verify_chain
from events import EVENTS, stamp
from attacker import KEY0, SIGNER, fresh_sealed, forge, key_at
from labkit import Clock, fresh_chain
from sealedlog import SealedWriter, verify_sealed

LATER = [(EVENTS[4], stamp(5)), (EVENTS[5], stamp(6))]


def noops(first, last):
    return [({"actor": "agent-7", "action": "noop", "target": str(i)}, stamp(i)) for i in range(first, last + 1)]


@pytest.fixture
def chain(tmp_path):
    fresh_chain(tmp_path / "chain.jsonl")
    return read_records(tmp_path / "chain.jsonl")


@pytest.fixture
def sealed(tmp_path):
    fresh_sealed(tmp_path / "sealed.jsonl")
    return read_records(tmp_path / "sealed.jsonl")


# ---- the hash chain (Step 2) ----

def test_clean_chain_verifies(chain):
    ok, message = verify_chain(chain)
    assert ok and message.startswith("6 entries")


@pytest.mark.parametrize("field", ["ts", "actor", "action", "target"])
def test_editing_any_field_is_detected(chain, field):
    chain[2][field] += "x"
    ok, message = verify_chain(chain)
    assert not ok and "hash mismatch" in message


@pytest.mark.parametrize("index", range(5))
def test_deleting_any_entry_but_the_last_is_detected(chain, index):
    del chain[index]
    assert not verify_chain(chain)[0]


def test_cutting_off_the_end_is_invisible_to_the_chain(chain):
    assert verify_chain(chain[:4])[0]


def test_swapping_neighbours_is_detected(chain):
    chain[1], chain[2] = chain[2], chain[1]
    assert not verify_chain(chain)[0]


def test_a_forged_insert_is_detected(chain):
    forged = dict(chain[3], seq=5, target="x")
    assert not verify_chain(chain[:4] + [forged] + chain[4:])[0]


def test_a_wrongly_linked_entry_with_a_valid_hash_is_detected(chain):
    chain[3] = build_record("f" * 64, 4, stamp(4), EVENTS[3])  # its own hash is right, its link is not
    ok, message = verify_chain(chain)
    assert not ok and "prev does not match" in message


def test_a_wrongly_numbered_last_entry_with_a_valid_hash_is_detected(chain):
    chain[5] = build_record(chain[4]["hash"], 9, stamp(6), EVENTS[5])  # right hash, right link, wrong number
    ok, message = verify_chain(chain)
    assert not ok and "expected seq 6" in message


def test_an_unkeyed_chain_can_be_rebuilt_by_anyone(chain):
    rebuilt, prev = [], GENESIS
    for number, record in enumerate([r for r in chain if r["seq"] not in (3, 4)], start=1):
        rebuilt.append(build_record(prev, number, record["ts"], {k: record[k] for k in ("actor", "action", "target")}))
        prev = rebuilt[-1]["hash"]
    assert verify_chain(rebuilt)[0]


# ---- the file format ----

@pytest.mark.parametrize("name", sorted(RESERVED))
def test_events_cannot_use_reserved_names(name):
    with pytest.raises(ValueError):
        build_record(GENESIS, 1, stamp(1), {name: "x"})


def test_strict_reader_rejects_a_duplicate_key(chain):
    text = line(chain[0]).rstrip("\n").replace('{"action"', '{"action":"read_file","action"', 1)
    assert json.loads(text) == chain[0]  # a lenient parser sees nothing wrong
    with pytest.raises(LogFormatError):
        parse_line(text)


def test_strict_reader_rejects_extra_whitespace(chain):
    with pytest.raises(LogFormatError):
        parse_line(json.dumps(chain[0], sort_keys=True))


def test_strict_reader_rejects_nan():
    with pytest.raises(ValueError):
        parse_line('{"seq":NaN}')


def test_non_ascii_events_round_trip(tmp_path):
    path = tmp_path / "unicode.jsonl"
    ChainWriter(path).append({"actor": "a", "action": "say", "target": "caf" + chr(0xE9) + chr(0x4E2D)}, stamp(1))
    assert path.read_bytes().isascii()
    assert read_records(path)[0]["target"] == "caf" + chr(0xE9) + chr(0x4E2D)
    assert verify_chain(read_records(path))[0]


def test_a_writer_resumes_from_the_file(tmp_path):
    fresh_chain(tmp_path / "a.jsonl")
    ChainWriter(tmp_path / "a.jsonl").append(EVENTS[0], stamp(7))
    ok, message = verify_chain(read_records(tmp_path / "a.jsonl"))
    assert ok and message.startswith("7 entries")


# ---- keys and key evolution (Steps 4 and 5) ----

@pytest.mark.parametrize("evolve_keys", [False, True])
def test_a_sealed_log_verifies(tmp_path, evolve_keys):
    fresh_sealed(tmp_path / "s.jsonl", evolve_keys)
    assert verify_sealed(read_records(tmp_path / "s.jsonl"), KEY0, evolve_keys)[0]


def test_a_wrong_key_fails_at_the_first_entry(sealed):
    ok, message = verify_sealed(sealed, b"\x01" * 32)
    assert not ok and message.startswith("seq 1: bad mac")


def test_a_flipped_mac_is_detected(sealed):
    mac = sealed[3]["mac"]
    sealed[3]["mac"] = ("0" if mac[0] != "0" else "1") + mac[1:]
    assert verify_sealed(sealed, KEY0)[1].startswith("seq 4: bad mac")


@pytest.fixture
def static(tmp_path):
    fresh_sealed(tmp_path / "static.jsonl", evolve_keys=False)
    return read_records(tmp_path / "static.jsonl")


def test_a_rebuild_without_the_key_fails(static):
    bad = forge(static[:2], LATER, b"\x00" * 32, evolve_keys=False)
    assert verify_sealed(bad, KEY0, evolve_keys=False) == (False, "seq 3: bad mac (not sealed by the key holder)")


def test_a_stolen_static_key_allows_a_full_rewrite(static):
    assert verify_sealed(forge(static[:2], LATER, KEY0, evolve_keys=False), KEY0, evolve_keys=False)[0]


@pytest.mark.parametrize("stolen_for,rewrite_from", [(3, 1), (4, 3), (7, 3), (7, 6)])
def test_a_stolen_evolving_key_cannot_rewrite_earlier_entries(sealed, stolen_for, rewrite_from):
    forged = forge(sealed[:rewrite_from - 1], noops(rewrite_from, 6), key_at(stolen_for))
    ok, message = verify_sealed(forged, KEY0)
    assert not ok and message.startswith(f"seq {rewrite_from}: bad mac")


@pytest.mark.parametrize("stolen_for", [1, 3, 5, 7])
def test_a_stolen_evolving_key_can_seal_everything_from_its_own_entry_on(sealed, stolen_for):
    forged = forge(sealed[:stolen_for - 1], noops(stolen_for, 8), key_at(stolen_for))
    assert verify_sealed(forged, KEY0)[0]


def test_the_writer_forgets_the_key_that_sealed_an_entry(tmp_path):
    writer = fresh_sealed(tmp_path / "s.jsonl")
    assert writer.current_key() == key_at(7)
    assert writer.current_key() not in {key_at(n) for n in range(1, 7)}


# ---- checkpoints (Step 6) ----

@pytest.fixture
def anchored(sealed):
    return sealed, make_checkpoint(SIGNER, "agent-actions", sealed[3]), SIGNER.public_key()


def test_a_checkpoint_holds_on_the_intact_log(anchored):
    records, checkpoint, public = anchored
    assert check_checkpoint(public, checkpoint, records)[0]


def test_cutting_below_the_checkpoint_is_detected(anchored):
    records, checkpoint, public = anchored
    ok, message = check_checkpoint(public, checkpoint, records[:3])
    assert not ok and "cut off" in message


def test_cutting_after_the_checkpoint_is_not_detected(anchored):
    records, checkpoint, public = anchored  # entries 5 and 6 are the exposure window
    assert check_checkpoint(public, checkpoint, records[:5])[0] and check_checkpoint(public, checkpoint, records[:4])[0]


def test_a_tail_rewrite_with_a_stolen_key_is_caught_by_the_checkpoint(anchored):
    records, checkpoint, public = anchored
    forged = forge(records[:2], noops(3, 6), key_at(3))
    assert verify_sealed(forged, KEY0)[0]
    ok, message = check_checkpoint(public, checkpoint, forged)
    assert not ok and "rewritten" in message


def test_a_checkpoint_signed_by_someone_else_is_rejected(anchored):
    records, _, public = anchored
    fake = make_checkpoint(Ed25519PrivateKey.generate(), "agent-actions", records[3])
    assert check_checkpoint(public, fake, records) == (False, "checkpoint signature is not valid")


def test_editing_a_checkpoint_breaks_its_signature(anchored):
    records, checkpoint, public = anchored
    assert not check_checkpoint(public, dict(checkpoint, seq=2), records)[0]


# ---- SQLite (Step 7) ----

@pytest.fixture
def db(tmp_path):
    connection = sqlitelog.connect(str(tmp_path / "agent.db"))
    for n, event in enumerate(EVENTS, start=1):
        sqlitelog.append(connection, event, stamp(n))
    yield connection
    connection.close()


def test_the_sqlite_log_verifies(db):
    assert verify_chain(sqlitelog.records(db))[0]


@pytest.mark.parametrize("statement", ["UPDATE log SET body = 'x' WHERE seq = 3", "DELETE FROM log WHERE seq = 3"])
def test_triggers_block_edits_through_the_application(db, statement):
    with pytest.raises(sqlite3.Error, match="append-only"):
        db.execute(statement)


def test_dropping_the_trigger_and_editing_is_caught_by_the_chain(db, tmp_path):
    raw = sqlite3.connect(str(tmp_path / "agent.db"), isolation_level=None)
    raw.execute("DROP TRIGGER log_no_update")
    raw.execute("UPDATE log SET body = replace(body, '.env', 'notes.txt') WHERE seq = 3")
    raw.close()
    ok, message = verify_chain(sqlitelog.records(db))
    assert not ok and message.startswith("seq 3: hash mismatch")


def test_four_processes_never_lose_or_fork_entries(tmp_path):
    path = str(tmp_path / "race.db")
    sqlitelog.connect(path).close()
    with multiprocessing.Pool(4) as pool:
        failed = pool.starmap(sqlitelog.append_many, [(path, who, 15, True) for who in range(1, 5)])
    check = sqlitelog.connect(path)
    records = sqlitelog.records(check)
    check.close()
    assert sum(failed) == 0 and len(records) == 60 and verify_chain(records)[0]


# ---- the agent hook (Step 9) ----

def test_the_call_is_logged_before_the_tool_runs(tmp_path):
    path = tmp_path / "hook.jsonl"
    writer, seen = SealedWriter(path, KEY0), []

    @audited(writer, Clock())
    def tool():
        seen.append([r["action"] for r in read_records(path)])

    tool()
    assert seen == [["tool.call"]]
    assert [r["action"] for r in read_records(path)] == ["tool.call", "tool.ok"]


def test_a_failing_tool_is_logged_and_the_error_still_propagates(tmp_path):
    path = tmp_path / "hook.jsonl"
    writer = SealedWriter(path, KEY0)

    @audited(writer, Clock())
    def tool():
        raise ConnectionError("refused")

    with pytest.raises(ConnectionError):
        tool()
    records = read_records(path)
    assert [r["action"] for r in records] == ["tool.call", "tool.error"] and records[1]["target"] == "ConnectionError"
    assert verify_sealed(records, KEY0)[0]

python -m pytest -q -p no:cacheprovider

......................................................                                                                                                             [100%]
54 passed in 1.17s

Passing tests only show that the code agrees with the tests. To see whether the tests would notice a bug, plant one. The script below copies the code to a temporary folder, makes one small change to a check, runs the suite and reports whether anything failed. It runs the suite once first with nothing changed, so a failure can only come from the planted bug. Save this as mutations.py.

# mutations.py
"""Plant one bug at a time in a copy of the code and check that the tests notice."""
import pathlib
import shutil
import subprocess
import sys
import tempfile

HERE = pathlib.Path(__file__).resolve().parent
SKIP = ("mutations", "run_all", "bench", "demo", "crash")
FILES = [p.name for p in HERE.glob("*.py") if not p.name.startswith(SKIP)]

MUTANTS = [
    ("verify_chain skips the prev check", "chainlog.py", 'if record.get("prev") != prev:', "if False:"),
    ("verify_chain skips the seq check", "chainlog.py", 'if record.get("seq") != index:', "if False:"),
    ("verify_chain skips the hash check", "chainlog.py", 'if record.get("hash") != entry_hash(record):', "if False:"),
    ("the hash leaves out the target field", "chainlog.py", 'if key not in ("hash", "mac")}', 'if key not in ("hash", "mac", "target")}'),
    ("the strict reader accepts any spelling", "chainlog.py", 'if not isinstance(record, dict) or text != canonical(record).decode("ascii"):', "if not isinstance(record, dict):"),
    ("verify_sealed skips the mac check", "sealedlog.py", 'if not hmac.compare_digest(str(record.get("mac", "")).encode(), expected.encode()):', "if False:"),
    ("verify_sealed never evolves the key", "sealedlog.py", "            key = evolve(key)", "            pass"),
    ("the writer never evolves its key", "sealedlog.py", "self._key[:] = evolve(bytes(self._key))  # the key that sealed this entry is gone after this line", "pass"),
    ("the checkpoint signature is not checked", "anchor.py", 'public_key.verify(bytes.fromhex(checkpoint["sig"]), DOMAIN + canonical(body))', "pass"),
    ("the checkpoint head is not compared", "anchor.py", 'if records[body["seq"] - 1]["hash"] != body["head"]:', "if False:"),
    ("the checkpoint length is not compared", "anchor.py", 'if len(records) < body["seq"]:', "if False:"),
    ("the hook forgets to log errors", "agent_hook.py", 'writer.append({**note, "action": f"{tool.__name__}.error", "target": type(error).__name__}, clock())', "pass"),
]


def run_tests(folder):
    command = [sys.executable, "-m", "pytest", "-q", "-x", "-p", "no:cacheprovider", "-k", "not four_processes"]
    result = subprocess.run(command, cwd=folder, capture_output=True, text=True, env={**__import__("os").environ, "PYTHONDONTWRITEBYTECODE": "1"})
    return result.returncode, [ln for ln in result.stdout.splitlines() if ln.strip()][-1]


def main():
    caught = 0
    with tempfile.TemporaryDirectory() as base:
        for label, filename, old, new in [("baseline, nothing planted", None, None, None)] + MUTANTS:
            folder = pathlib.Path(base) / "copy"
            shutil.rmtree(folder, ignore_errors=True)
            folder.mkdir()
            for name in FILES:
                shutil.copy(HERE / name, folder / name)
            if filename:
                path = folder / filename
                text = path.read_text(encoding="utf-8")
                assert text.count(old) == 1, (label, text.count(old))
                path.write_text(text.replace(old, new), encoding="utf-8", newline="\n")
            code, last = run_tests(folder)
            if filename is None:
                print(f"{'baseline':<9} {label:<46} {last}")
                continue
            caught += code != 0
            print(f"{'caught' if code else 'SURVIVED':<9} {label:<46} {last}")
    print(f"\n{caught} of {len(MUTANTS)} planted bugs were caught")


if __name__ == "__main__":
    main()

python mutations.py

baseline  baseline, nothing planted                      53 passed, 1 deselected in 0.68s
caught    verify_chain skips the prev check              1 failed, 13 passed, 1 deselected in 0.47s
caught    verify_chain skips the seq check               1 failed, 14 passed, 1 deselected in 0.26s
caught    verify_chain skips the hash check              1 failed, 1 passed, 1 deselected in 0.15s
caught    the hash leaves out the target field           1 failed, 4 passed, 1 deselected in 0.14s
caught    the strict reader accepts any spelling         1 failed, 21 passed, 1 deselected in 0.27s
caught    verify_sealed skips the mac check              1 failed, 28 passed, 1 deselected in 0.32s
caught    verify_sealed never evolves the key            1 failed, 27 passed, 1 deselected in 0.33s
caught    the writer never evolves its key               1 failed, 27 passed, 1 deselected in 0.31s
caught    the checkpoint signature is not checked        1 failed, 45 passed, 1 deselected in 0.51s
caught    the checkpoint head is not compared            1 failed, 44 passed, 1 deselected in 0.53s
caught    the checkpoint length is not compared          1 failed, 42 passed, 1 deselected in 0.48s
caught    the hook forgets to log errors                 1 failed, 52 passed, 1 deselected in 0.70s

12 of 12 planted bugs were caught

All 12 planted bugs were caught, including the quiet ones: leaving the target field out of the hash, accepting any spelling of a line, never evolving the key, and forgetting to compare the checkpoint’s hash. Only the multi-process test is skipped here, because it takes longer and none of the mutants touch it.

Step 10: Measure what the protection costs

Do hash chains and key evolution slow an agent down? Measure instead of guessing. This script seals 100,000 entries in memory, verifies them, then compares three ways of writing a line to disk: with the file kept open, with an fsync after every entry, and with the file opened and closed for every entry as the writers in this tutorial do. Finally it times how long a writer takes to start on a 100,000-entry file. Save this as bench.py.

# bench.py
"""Step 10: what does the protection cost? Sealing and verification speed, three ways of writing, and writer start-up time."""
import os
import tempfile
import time

from chainlog import line, read_records, verify_chain
from events import EVENTS
from labkit import KEY0, forge
from sealedlog import SealedWriter, verify_sealed

TS = "2026-10-30T09:00:00Z"
EVENT_LIST = [(EVENTS[i % len(EVENTS)], TS) for i in range(100000)]


def timed(label, count, work, unit="entries"):
    start = time.perf_counter()
    work()
    seconds = time.perf_counter() - start
    print(f"{label:<62}{count / seconds:>10,.0f} {unit}/s ({seconds / count * 1000:.3f} ms each)")


box = {}
timed("seal 100,000 entries in memory (hash, MAC, key evolution)", 100000, lambda: box.update(records=forge([], EVENT_LIST, KEY0)))
records = box["records"]
timed("verify the chain only", 100000, lambda: verify_chain(records))
timed("verify the chain, every MAC and the key evolution", 100000, lambda: verify_sealed(records, KEY0))

with tempfile.TemporaryDirectory() as folder:
    path = os.path.join(folder, "log.jsonl")

    def kept_open(fsync):
        with open(path, "w", encoding="ascii", newline="\n") as handle:
            for record in records[:500 if fsync else 20000]:
                handle.write(line(record))
                handle.flush()
                if fsync:
                    os.fsync(handle.fileno())

    def open_each_time():
        with open(path, "w"):
            pass
        for record in records[:500]:
            with open(path, "a", encoding="ascii", newline="\n") as handle:
                handle.write(line(record))

    timed("write, file kept open, flush after each entry", 20000, lambda: kept_open(False))
    timed("write, file kept open, flush and fsync after each entry", 500, lambda: kept_open(True))
    timed("write, open and close the file for every entry (our writers)", 500, open_each_time)

    with open(path, "w", encoding="ascii", newline="\n") as handle:
        handle.write("".join(line(record) for record in records))
    size = os.path.getsize(path)
    start = time.perf_counter()
    SealedWriter(path, KEY0)
    print(f"\nopening a writer on a {len(records):,} entry file ({size / len(records):.0f} bytes per entry, {size / 1e6:.1f} MB): {time.perf_counter() - start:.2f} s")

python bench.py

seal 100,000 entries in memory (hash, MAC, key evolution)        198,592 entries/s (0.005 ms each)
verify the chain only                                            380,728 entries/s (0.003 ms each)
verify the chain, every MAC and the key evolution                227,659 entries/s (0.004 ms each)
write, file kept open, flush after each entry                     51,694 entries/s (0.019 ms each)
write, file kept open, flush and fsync after each entry            1,264 entries/s (0.791 ms each)
write, open and close the file for every entry (our writers)         174 entries/s (5.743 ms each)

opening a writer on a 100,000 entry file (334 bytes per entry, 33.4 MB): 0.58 s

Your numbers will differ, and these came from one Windows 11 machine, but the shape should carry over:

• The cryptography is cheap. Hashing, MACing and evolving the key cost about 5 microseconds per entry, and verification runs at roughly 230,000 entries a second with MACs and key evolution.
• The disk is the cost. Flushing after each entry on an open file ran at about 52,000 entries a second, adding an fsync dropped that to about 1,300, and reopening the file for every entry, which is what the simple writers here do, reached only about 170. I did not investigate why opening and closing a file is so slow on this machine; the point is that it dwarfs the hashing by three orders of magnitude.
• Start-up reads the whole file. Opening a writer on a 100,000-entry file (334 bytes per entry, 33.4 MB) took 0.58 seconds because it parses every line. For a log that grows without end, keep the latest sequence number and hash in a small separate file, or read only the tail, and let a scheduled job do the full verification.

An agent making a tool call every few seconds is nowhere near any of these limits. If you do have a high volume, keep the file open, flush on every entry and call fsync on a timer or every few hundred entries, and accept that a power cut can lose the entries since the last fsync.

What each layer catches

This table collects what the demos showed. Caught means the verification step you built reported it; missed means it passed.

No row is caught in every column, and the last row is missed in all of them. That is the honest shape of the problem: each layer narrows what an attacker can do undetected, and the last line of defence is a place the attacker cannot reach.

Common mistakes and gotchas

• Verifying on the host that writes the log. The verifier needs its own copy of the initial key and the public signing key, on a different machine, and it should run on a schedule and alert on a FAIL. A check nobody runs catches nothing.
• Hashing the object instead of the bytes. Re-serializing parsed data before hashing means two different files can verify the same. Store the canonical bytes and compare them, as parse_line does.
• Keeping checkpoints next to the log. An attacker who can delete the log can delete the checkpoints. Store them elsewhere and remember the newest one your verifier has accepted.
• Logging after the action. A crash then leaves an action with no entry. Log the call first, as in Step 8.
• Putting secrets in entries. You cannot redact an entry later without breaking the chain. Log references and salted hashes.
• Two writers on one chain. Use one logging service, or take the database write lock before reading the head, as in Step 7.
• Believing the timestamp. The chain proves order, not time. The ts field is whatever the writer claimed. A trusted time comes from the place that stores the checkpoints, which can stamp them on arrival.
• Treating the log as proof of truth. It proves the record was not changed after it was written. An agent that lies to its own logger, or an action that never reaches the logger, is outside what this design can show.

Check that everything works end to end

With all the files in one folder, the script below runs every demo, the tests, the mutation check and the benchmark, and saves each output to an out folder. Save it as run_all.py.

# run_all.py
"""Run every step, the tests, the mutation check and the benchmark, saving each output to out/NAME.txt."""
import os
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "out")
os.makedirs(OUT, exist_ok=True)
STEPS = [
    "demo1_plain", "demo2_attacks", "demo3_rechain", "demo4_hmac", "demo5_forward",
    "demo6_anchor", "demo7_sqlite", "demo8_agent", "pytest", "mutations", "bench",
]
only = sys.argv[1:]
env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", PYTHONIOENCODING="utf-8", COLUMNS="170")
for name in STEPS:
    if (only and name not in only) or (not only and name == "bench" and os.environ.get("SIM_SKIP_BENCH")):
        continue
    command = [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider"] if name == "pytest" else [sys.executable, f"{name}.py"]
    start = time.time()
    result = subprocess.run(command, cwd=HERE, env=env, capture_output=True, text=True, encoding="utf-8", errors="replace")
    text = result.stdout + (("\n[stderr]\n" + result.stderr) if result.stderr.strip() and name != "pytest" else "")
    with open(os.path.join(OUT, f"{name}.txt"), "w", encoding="utf-8", newline="\n") as handle:
        handle.write(text)
    print(f"{name:16} exit {result.returncode} in {time.time() - start:5.1f} s, {len(text)} chars")

python run_all.py

You should see one line per step, each reporting exit 0. Then open out/pytest.txt and look for 54 passed, and open out/mutations.txt and look for 12 of 12 planted bugs caught. The demo outputs should match this page line for line, apart from the lines marked as varying and the benchmark timings. If a hash differs, the usual cause is a changed timestamp or a stray space in an event, because every byte of a record goes into its hash.

Where to go next

• Prove one entry without sending the whole log. A hash chain forces an auditor to read everything. Crosby and Wallach’s paper Efficient Data Structures for Tamper-Evident Logging says: “Where a classic hash chain might require an 800 MB trace to prove that a randomly chosen event is in a log with 80 million events, our prototype returns a 3 KB proof with the same semantics.” The structure behind that is a tree. Our Merkle tree tutorial builds the same family of structure, and RFC 9162 describes the version Certificate Transparency uses.
• Make appends crash-safe. The write-ahead log tutorial covers torn writes and recovery, which this tutorial left out: a half-written last line makes read_records raise, and you need a policy for it.
• Move the keys out of the process. Put the initial key and the signing key in a secret manager or a hardware security module, and let a small logging service be the only holder of the evolving key.
• Wire it into a real agent. The agent harness tutorial has the tool loop and the JSONL event log; swap its writer for SealedWriter and its tool wrappers for audited.

Sources and further reading

• Bruce Schneier and John Kelsey, Secure Audit Logs to Support Computer Forensics: the key evolution design and the abnormal-shutdown discussion quoted above.
• Scott A. Crosby and Dan S. Wallach, Efficient Data Structures for Tamper-Evident Logging.
• AWS, Validating CloudTrail log file integrity.
• systemd, journald.conf, the Seal= option.
• T. Bray, RFC 8259, The JSON Data Interchange Format.
• SQLite, BEGIN TRANSACTION.
• Python, os.fsync.