examples/basic_log.py
1.8 KB · 67 lines · python Raw
1 """basic_log.py - write 30 events with rotation, then verify the chain.
2
3 Run::
4
5 python examples/basic_log.py
6 """
7
8 from __future__ import annotations
9
10 import tempfile
11
12 from quantumshield.identity.agent import AgentIdentity
13
14 from pqc_audit_log_fs import (
15 InferenceEvent,
16 LogAppender,
17 LogReader,
18 RotationPolicy,
19 )
20
21
22 def main() -> None:
23 with tempfile.TemporaryDirectory() as log_dir:
24 signer = AgentIdentity.create(name="demo-signer")
25 appender = LogAppender(
26 log_dir,
27 signer,
28 rotation=RotationPolicy(max_events_per_segment=10),
29 )
30
31 for i in range(30):
32 event = InferenceEvent.create(
33 model_did="did:pqaid:demo-model",
34 model_version="1.0.0",
35 input_bytes=f"input-{i}".encode(),
36 output_bytes=f"output-{i}".encode(),
37 decision_type="classification",
38 decision_label="approve" if i % 2 == 0 else "deny",
39 actor_did="did:pqaid:demo-user",
40 session_id=f"sess-{i // 5}",
41 )
42 appender.append(event)
43 appender.close()
44
45 reader = LogReader(log_dir)
46 segments = reader.list_segments()
47 print(f"wrote {len(segments)} sealed segments")
48 for n in segments:
49 header = reader.read_header(n)
50 print(
51 f" segment {n:05d} "
52 f"events={header.event_count} "
53 f"root={header.merkle_root[:16]}..."
54 )
55
56 ok, errors = reader.verify_chain()
57 if ok:
58 print(f"[OK] chain verifies across {len(segments)} segments")
59 else:
60 print("[FAIL] chain verification failed:")
61 for e in errors:
62 print(f" - {e}")
63
64
65 if __name__ == "__main__":
66 main()
67