This module sets up a dual-handler logger: a console handler for humans and a file handler that writes one JSON object per line, conforming to the schema at audit_log_schema.json next to this file. The file is the compliance-relevant sink (EU AI Act Article 12, IEC 62443-4-2 SAR 6.1, IEC 61508-3 §7.4.7); the console handler is plaintext for interactive use.
Schema versioning rule: SCHEMA_VERSION is loaded directly from audit_log_schema.json at import time, so there is exactly one source of truth. Any change to the schema file is a breaking change to the public compliance contract and MUST be released as a MAJOR version, with a Conventional-Commits breaking-change marker (type!:) on at least one commit so that it appears as such in the changelog. This was previously enforced by the audit-log-schema-gate CI job; since the move to a local release workflow it is a maintainer obligation with no automated gate.
Integrity: every record after the first carries prev_hash, the SHA-256 hex digest of the exact bytes written for the previous record’s line (line terminator excluded); the first record of a file carries the genesis value GENESIS_HASH (64 "0" characters). The chain is per file — a new timestamp-named file starts a new chain, log rotation is unsupported — and it is unkeyed: it proves that a retained file was not edited, deleted, reordered, or torn after the fact, not who wrote it. Verify a file offline with verify_audit_log, which recomputes the chain from scratch and reports the first broken line, the record count, and — when the caller presents an AuditAnchor obtained out of band — whether this file’s chain still extends the anchored state.
verify_audit_log reports anchoring status explicitly rather than defaulting to a clean result: AnchorStatus.NOT_PRESENTED is never treated as success, and every AuditChainReport names exactly what could not be established (AuditChainReport.unestablished) alongside what could — an unanchored result is never printed as an unqualified pass. Binding head_hash to an independently controlled record still happens outside this module: make anchor / make anchor-verify (scripts/anchor_audit_log.py) automate the round trip to an RFC 3161 timestamping authority, but the library itself performs no network I/O — AuditAnchor and AuditAnchorSet only compare an already-obtained claim against a chain recomputed offline. See docs/anchoring.qmd for the operator runbook and docs/security.qmd for the residual risk this leaves.
Format LogRecord instances as single-line JSON per audit_log_schema.json.
AnchorStatus
manager.logger.AnchorStatus()
Where a presented AuditAnchor’s head hash was found in a file’s chain.
verify_audit_log never defaults this to a success value: when no anchor is passed the status is NOT_PRESENTED, not some implicit “not applicable, therefore fine”.
An externally witnessed claim about one audit log file’s chain head.
An AuditAnchor is operator input, not something this module produces: it is whatever an external witness (an RFC 3161 timestamping authority, a second organisation’s log, a notarised printout, …) attested about a file’s AuditChainReport.head_hash at some point in the past. Everything except head_hash is unauthenticated operator input — n_records, anchored_at, witness, and log_file are carried into the AuditChainReport so a printed result says where the claim came from, but verify_audit_log never trusts them for the verdict itself; only head_hash is ever compared against the recomputed chain. Verifying an RFC 3161 token over to_json() is a separate, external step — scripts/anchor_audit_log.py drives it via openssl ts, this module performs no network I/O and parses no timestamp tokens.
Validation is strict, not normalising: head_hash must already be exactly 64 lowercase hex characters with no surrounding whitespace, and must not be GENESIS_HASH (identical for every empty-or-fresh log on earth, so it anchors nothing). anchored_at, when given, must be timezone-aware; a naive datetime is ambiguous about which instant it names and is rejected rather than assumed to be UTC.
The record count the witness observed at anchoring time, if recorded. Used only to flag an anchor whose own bookkeeping is self-inconsistent (head_hash matches a different position than the one n_records claims).
The audit log file name this anchor is about. Required inside an AuditAnchorSet; optional standalone, since a bare anchor is usually already scoped to one file by the caller.
Strict validation rejects a well-formed-looking but ambiguous input rather than silently normalising it:
from spotforecast2_safe.manager.logger import GENESIS_HASH, AuditAnchorfor bad_hash in ("A"*64, "1"*64+" ", GENESIS_HASH):try: AuditAnchor(head_hash=bad_hash)exceptValueErroras exc:print(f"rejected {bad_hash[:12]!r}...: {exc}")
rejected 'AAAAAAAAAAAA'...: head_hash must be exactly 64 lowercase hex characters with no surrounding whitespace; got 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
rejected '111111111111'...: head_hash must be exactly 64 lowercase hex characters with no surrounding whitespace; got '1111111111111111111111111111111111111111111111111111111111111111 '
rejected '000000000000'...: head_hash is the genesis value (64 zeros), which is identical for every audit log on earth and anchors nothing
If text is not a JSON object, declares an anchor_schema_version other than ANCHOR_SCHEMA_VERSION, carries an unrecognised top-level key, or is missing head_hash.
to_json
manager.logger.AuditAnchor.to_json()
Serialise this anchor as a canonical JSON document.
One canonical document anchoring every retained audit log file at once.
setup_logging writes one file per process start, so a single per-file AuditAnchor goes stale the moment a second process starts — it says nothing about the new file. AuditAnchorSet closes that gap by naming every file the operator intends to retain, together with its (log_file, n_records, head_hash). A file present on disk but absent from the most recent set, or named in the set but missing from disk, is itself a finding (the latter is the Article 18 retention gap; scripts/anchor_audit_log.py --log-dir ... --verify reports both).
Every member AuditAnchor must carry AuditAnchor.log_file, and file names must be unique within the set — both are enforced at construction. Verification itself stays per-file: verify_audit_log takes one AuditAnchor, never a set; for_log_file is the lookup for a caller that needs the member anchor for a single named file.
rejected duplicate: duplicate log_file name(s) in AuditAnchorSet: ['a.log']
rejected unnamed anchor: every anchor in an AuditAnchorSet must carry log_file
If text is not a JSON object, declares an anchor_schema_version other than ANCHOR_SCHEMA_VERSION, carries an unrecognised key at the top level or within an entry, or an entry is missing head_hash.
to_json
manager.logger.AuditAnchorSet.to_json()
Serialise the whole set as one canonical JSON document.
AuditCaveat
manager.logger.AuditCaveat()
A named gap in what one verify_audit_log run has established.
The first six members are CONTINGENT on the file and the anchor presented for a particular run; they populate AuditChainReport.unestablished. The last two, AUTHORSHIP_NOT_ESTABLISHED and SCOPE_LIMITED_TO_THIS_FILE, are STANDING — true of every run regardless of outcome — and are never placed in unestablished; they live only in STANDING_LIMITATIONS and are rendered by AuditChainReport.summary.
True when every record’s prev_hash matches the SHA-256 of the previous record’s bytes, from GENESIS_HASH through the last line. True for an empty file (vacuously intact). Scoped to exactly that: it says nothing about whether the file’s current state was ever witnessed outside itself — that is what anchor_status is for.
Number of records in the verified, intact prefix. Equal to the file’s total line count when chain_intact is True; equal to the count of well-formed records before the break otherwise.
Human-readable diagnostic for the first problem found: a chain break (paired with first_broken_line), or — when the chain is otherwise intact but a presented anchor’s own AuditAnchor.n_records is inconsistent with the position where its AuditAnchor.head_hash actually matched — an explanation of that inconsistency. None when neither applies. Not a stable API: programmatic callers branch on chain_intact and anchor_status, never on the text of reason.
The prev_hash the next record appended to this file must carry to extend the chain. Never None: GENESIS_HASH for an empty file, otherwise the SHA-256 of the last line of the verified intact prefix — including when chain_intact is False, so an operator can tell exactly where a repaired chain would have to resume.
Where the presented anchor’s head hash was found, or AnchorStatus.NOT_PRESENTED when verify_audit_log was called without one. Never defaults to a success value.
The prefix length at which the presented anchor’s head hash was found (None when anchor_status is AnchorStatus.NOT_PRESENTED or AnchorStatus.MISMATCH).
The CONTINGENT AuditCaveat values that apply to this run, in AuditCaveat declaration order. Empty only when chain_intact and anchor_status is AnchorStatus.MATCHES_HEAD and n_records > 0 — i.e. not report.unestablished is a reachable, honest predicate, but only ever true for a run that presented a head-matching anchor. The two STANDING caveats (STANDING_LIMITATIONS) are never in this tuple; they are true of every run and are rendered separately by summary.
audit chain BROKEN at line 3 (record 3 carries prev_hash '11' * 32, expected sha256(record 2)); 2 records verified, prefix head 00000000…; NOT ESTABLISHED: external anchoring (a wholesale rewrite of this file by whoever holds it cannot be excluded), records after the break, final record not pinned
Prints the file path, the chain verdict, the full (untruncated) head hash, the anchor block (including its unauthenticated metadata, clearly labelled as such), a “NOT ESTABLISHED (this file)” list of the contingent caveats that apply to this run, and a “STANDING LIMITATIONS” list of the two caveats true of every run. Every caveat is rendered as a full sentence.
Format LogRecord instances as single-line JSON per audit_log_schema.json.
The formatter emits exactly the fields named in the schema’s properties section, never more. Callers pass optional structured context through the standard loggingextra= mechanism; recognised extras are event, task, and context.
Each instance also hash-chains the records it emits: it holds the SHA-256 of the previous formatted line in self._prev_hash (seeded to GENESIS_HASH) and stamps it into the next record’s prev_hash field, then advances the state to the hash of the line it just returned. This makes one formatter instance per sink a hard invariant, not a convenience: two handlers sharing one formatter interleave into a single chain, so tampering with either sink’s file looks like tampering with both (verify_audit_log reports a break at the first line either sink wrote out of turn). setup_logging respects this by constructing a fresh formatter per file handler.
Two attachment patterns silently break the chain and are out of scope: a logging.handlers.QueueHandler or logging.handlers.MemoryHandler in front of this formatter, because both call format() to build their queued/buffered record without writing it, advancing _prev_hash for a line that may never reach disk; and a logging.handlers.RotatingFileHandler, because rotation starts a new file mid-chain with no genesis record.
Examples
import ioimport jsonimport loggingfrom spotforecast2_safe.manager.logger import ( GENESIS_HASH, JsonAuditFormatter, SCHEMA_VERSION,)formatter = JsonAuditFormatter()stream = io.StringIO()handler = logging.StreamHandler(stream)handler.setFormatter(formatter)logger = logging.getLogger("example.audit")logger.setLevel(logging.DEBUG)# Avoid duplicate handlers across repeated runs in the same processlogger.handlers.clear()logger.addHandler(handler)logger.propagate =Falselogger.info("model fitted", extra={"event": "fit", "task": "demo", "context": {"lags": 3}},)line = stream.getvalue().strip()record = json.loads(line)assert record["schema_version"] == SCHEMA_VERSIONassert record["event"] =="fit"assert record["task"] =="demo"assert record["context"] == {"lags": 3}assert record["prev_hash"] == GENESIS_HASH # first record of the chainprint(f"schema_version={record['schema_version']} event={record['event']}")
Configure dual-handler logging for safety-critical execution.
Attaches a stream handler (stdout, human-readable plaintext) and, when log_dir is provided, a file handler that writes JSON records per audit_log_schema.json. The console handler honours level; the file handler is always at INFO so the audit trail stays complete even when the operator silences the console.
The file is opened with encoding="utf-8" (independent of locale) and mode="x": exclusive creation, so a second process that starts within the same wall-clock second cannot append a second hash chain onto one file — it gets an OSError instead. Exactly one file is created per process start, and each file carries its own self-contained SHA-256 hash chain seeded at GENESIS_HASH; verify it offline with verify_audit_log.
Optional directory for the audit log file. If provided, a timestamped sf2-safe-logger_YYYYMMDD_HHMMSS.log file is created and receives JSON-formatted, hash-chained records.
If log_dir is provided but the directory cannot be created, or the audit log file cannot be opened — including the case where a file of that name already exists, since the handler now opens with mode="x". In a safety-critical workflow a missing or ambiguous audit trail is not recoverable — the failure surfaces immediately rather than degrading silently to console-only logging or splicing two chains into one file.
Examples
import jsonimport loggingimport tempfilefrom pathlib import Pathfrom spotforecast2_safe.manager.logger import setup_logging, verify_audit_log# Reset the named logger so the example is idempotent when the notebook# kernel re-runs this cell.named = logging.getLogger("sf2-safe-logger")named.handlers.clear()with tempfile.TemporaryDirectory() as tmp: log_dir = Path(tmp) logger, log_path = setup_logging(level=logging.WARNING, log_dir=log_dir)assert log_path isnotNoneassert log_path.exists() logger.info("pipeline started", extra={"event": "task_start"}) lines = [l for l in log_path.read_text(encoding="utf-8").splitlines() if l]assertlen(lines) >=1 record = json.loads(lines[0])assert record["event"] =="audit_log_init"print(f"log file created: {log_path.name}")print(f"first record event: {record['event']}")for h in logger.handlers: h.flush() report = verify_audit_log(log_path)print(f"chain intact: {report.chain_intact}, n_records: {report.n_records}")# Tear down so subsequent cells start cleannamed.handlers.clear()
log file created: sf2-safe-logger_20260810_141649.log
first record event: audit_log_init
chain intact: True, n_records: 2
Recompute the SHA-256 hash chain of an audit log file and, optionally, compare it against an externally witnessed AuditAnchor.
Reads path in binary and, in order, walks every line: (1) decodes it as UTF-8; (2) parses it as a JSON object; (3) checks schema_version == SCHEMA_VERSION; (4) checks that prev_hash is present and a string; (5) checks that prev_hash equals the SHA-256 hex digest of the previous line’s bytes (GENESIS_HASH for line 1). The walk stops at the first failing check, so AuditChainReport.n_records counts only the verified, intact prefix.
Splitting happens on b"\n" only, matching binary file iteration and never str.splitlines(), so a stray U+2028/U+0085 inside a message field (left unescaped by json.dumps(..., ensure_ascii=False)) cannot masquerade as a record boundary. A final segment with no trailing newline (a torn last write) is reported as a broken line rather than silently dropped or repaired — this function never rewrites the file. One optional trailing b"\r" is stripped from each segment before decoding and hashing, so files written with CRLF line endings verify identically to LF ones.
When anchor is given (an AuditAnchor, or a bare 64-character lowercase hex string, which is wrapped into one with no other fields set), its AuditAnchor.head_hash is compared against every prefix head reached during the same walk — not only the final one. This matters for a log file that is still being appended to: comparing only against the final head would report AnchorStatus.MISMATCH for a file that has simply grown since the anchor was made, and a rail that cries wolf on every legitimate append gets switched off within a week. Instead: AnchorStatus.MATCHES_HEAD when the anchor’s hash equals the head of the entire verified prefix, AnchorStatus.MATCHES_PREFIX when it equals an earlier position, and AnchorStatus.MISMATCH when it is never found (including when AuditAnchor.n_records disagrees with the position where the hash did match — a swapped or edited sidecar). A broken chain does not suppress the comparison: AnchorStatus.MATCHES_PREFIX together with AuditChainReport.chain_intact being False is valid forensics — it says the file used to extend a witnessed state and was corrupted only after that point.
Documented non-guarantee: this comparison detects edits, deletions, and reordering inside a retained file relative to what was anchored, but it cannot detect a file truncated back to exactly its anchored state (that reports AnchorStatus.MATCHES_HEAD, correctly, since the remaining bytes really are unaltered), and it cannot detect deletion of records appended after the anchor except via a later anchor that would have covered them. Anchoring cadence is entirely the operator’s control — this function has no opinion on how often to call scripts/anchor_audit_log.py. Whole-file deletion is not observable from the file at all; see AuditAnchorSet for that case.
An externally witnessed claim about this file’s chain head, or None to verify the chain alone. None reports AnchorStatus.NOT_PRESENTED, never a default success value.
If path exists but cannot be read. Propagated unchanged, not wrapped: a missing or unreadable audit trail is a failure of the audit, not a verdict verify_audit_log can render — consistent with setup_logging’s fail-loud stance on audit-file errors.
Examples
import loggingimport tempfilefrom pathlib import Pathfrom spotforecast2_safe.manager.logger import setup_logging, verify_audit_log# Reset the named logger so the example is idempotent when the notebook# kernel re-runs this cell.named = logging.getLogger("sf2-safe-logger")named.handlers.clear()with tempfile.TemporaryDirectory() as tmp: log_dir = Path(tmp) logger, log_path = setup_logging(log_dir=log_dir) logger.info("fit complete", extra={"event": "fit"}) logger.info("predict complete", extra={"event": "predict"})for h in logger.handlers: h.flush() report = verify_audit_log(log_path)print(report)assert report.chain_intactassert report.n_records ==3# audit_log_init + the two calls aboveassert report.first_broken_line isNoneassertnot report.anchored # no anchor was presented# Tear down so subsequent cells start cleannamed.handlers.clear()
2026-08-10 14:16:49,881 - sf2-safe-logger - INFO - Persistent logging initialized at: /var/folders/dw/pvtj6mt91znd0hftcztqb0k00000gn/T/tmpota8z5ne/sf2-safe-logger_20260810_141649.log
2026-08-10 14:16:49,882 - sf2-safe-logger - INFO - fit complete
2026-08-10 14:16:49,882 - sf2-safe-logger - INFO - predict complete
audit chain intact: 3 records, head 8d8c90cf…; NOT ESTABLISHED: external anchoring (a wholesale rewrite of this file by whoever holds it cannot be excluded), final record not pinned
Presenting an anchor changes the verdict, entirely offline. This example builds a short chain by hand, anchors its head, then walks it through the three outcomes an operator will actually see over a file’s lifetime — matched, grown, and rewritten:
import loggingimport tempfilefrom pathlib import Pathfrom spotforecast2_safe.manager.logger import ( JsonAuditFormatter, verify_audit_log,)def _record(msg: str) -> logging.LogRecord:return logging.LogRecord( name="demo", level=logging.INFO, pathname="", lineno=0, msg=msg, args=(), exc_info=None, )formatter = JsonAuditFormatter()lines = [formatter.format(_record(f"event {i}")) for i inrange(3)]path = Path(tempfile.mkdtemp()) /"audit.log"path.write_text("\n".join(lines) +"\n", encoding="utf-8")anchored_head = verify_audit_log(path).head_hash# 1. Re-verify unchanged: the anchor matches the current head exactly.report = verify_audit_log(path, anchor=anchored_head)print(report)assert report.anchor_status.value =="matches_head"assertnot report.unestablished # only case where this is empty# 2. Append a new, correctly-chained record: the anchor still matches,# but only a prefix of the now-longer file.extra = formatter.format(_record("event 3"))withopen(path, "a", encoding="utf-8") as fh: fh.write(extra +"\n")grown_report = verify_audit_log(path, anchor=anchored_head)print(grown_report)assert grown_report.anchor_status.value =="matches_prefix"assert grown_report.anchored_records ==3# 3. Rewrite the file wholesale (a fresh chain, same anchor presented):# the anchor's head hash is not found anywhere in the new chain.fresh_formatter = JsonAuditFormatter() # a new chain starts at GENESIS_HASHrewritten = [fresh_formatter.format(_record(f"forged {i}")) for i inrange(3)]path.write_text("\n".join(rewritten) +"\n", encoding="utf-8")mismatch_report = verify_audit_log(path, anchor=anchored_head)print(mismatch_report)assert mismatch_report.anchor_status.value =="mismatch"assert mismatch_report.chain_intact # the forged chain is internally consistentassertnot mismatch_report.anchored
audit chain intact: 3 records, head 45eead93…; anchor MATCHES HEAD; standing limits: unkeyed (no authorship), scope is this file only
audit chain intact: 4 records, head 50e017e7…; NOT ESTABLISHED: records after the anchor, final record not pinned
audit chain intact: 3 records, head f636430a…; NOT ESTABLISHED: anchored state is not this file, final record not pinned