Create brutal_test.py

brutal_test.py

@@ -0,0 +1,125 @@
+import os, json, tempfile, shutil, zipfile
+import rft_flightrecorder as fr
+
+def read_lines(p):
+    with open(p, "r", encoding="utf-8") as f:
+        return [ln.rstrip("\n") for ln in f if ln.strip()]
+
+def write_lines(p, lines):
+    with open(p, "w", encoding="utf-8") as f:
+        for ln in lines:
+            f.write(ln + "\n")
+
+def assert_true(cond, msg):
+    if not cond:
+        raise AssertionError(msg)
+
+def assert_false(cond, msg):
+    if cond:
+        raise AssertionError(msg)
+
+def main():
+    td = tempfile.mkdtemp(prefix="rft_brutal_")
+    log = os.path.join(td, "flightlog.jsonl")
+    print("TEMP:", td)
+
+    sk, pk = fr.gen_keys()
+
+    # --- PASS flow ---
+    sid, msg = fr.start_session(log, "audit-demo", "deterministic", "brutal test", True, sk)
+    print("start:", msg)
+    assert_true(sid, "No session_id returned")
+
+    # append signed events
+    for i in range(1, 6):
+        ev, m = fr.append_event(
+            log_path=log,
+            session_id=sid,
+            event_type="note",
+            payload_text=json.dumps({"i": i, "msg": "ok"}, ensure_ascii=False),
+            parent_event_hash="",
+            sign_event=True,
+            sk_hex=sk,
+            model_id="audit-demo",
+            run_mode="deterministic",
+        )
+        assert_true(ev is not None, f"append failed at {i}: {m}")
+
+    status, ok, report = fr.verify_session(log, sid, pk, require_signatures=True)
+    print("verify signed:", status)
+    assert_true(ok, "Signed session should verify PASS")
+
+    anchor, msg = fr.finalise_session(log, sid, True, sk, "audit-demo", "deterministic")
+    print("finalise:", msg)
+    assert_true(anchor is not None, "Finalise failed")
+
+    status, ok, report = fr.verify_session(log, sid, pk, require_signatures=True)
+    print("verify after finalise:", status)
+    assert_true(ok, "Session should still verify after finalise")
+
+    bundle, msg = fr.export_session_bundle(log, sid)
+    print("export:", msg)
+    assert_true(bundle and os.path.exists(bundle), "Bundle not created")
+
+    # import verify
+    status, ok, report, stored = fr.import_bundle_verify(bundle, pk_hex=pk, require_signatures=False, store_into_log=False, log_path=log)
+    print("import verify:", status)
+    assert_true(ok, "Imported bundle should verify PASS")
+
+    # --- FAIL: tamper payload ---
+    lines = read_lines(log)
+    assert_true(len(lines) > 3, "Not enough log lines")
+
+    tampered = lines[:]
+    obj = json.loads(tampered[2])
+    obj["payload"]["msg"] = "hacked"
+    tampered[2] = json.dumps(obj, ensure_ascii=False)
+    tamper_log = os.path.join(td, "tamper_payload.jsonl")
+    write_lines(tamper_log, tampered)
+
+    status, ok, report = fr.verify_session(tamper_log, sid, pk, require_signatures=True)
+    print("tamper payload verify:", status)
+    assert_false(ok, "Tampered payload must FAIL")
+
+    # --- FAIL: reorder lines ---
+    reordered = lines[:]
+    if len(reordered) >= 4:
+        reordered[2], reordered[3] = reordered[3], reordered[2]
+    reorder_log = os.path.join(td, "tamper_reorder.jsonl")
+    write_lines(reorder_log, reordered)
+
+    status, ok, report = fr.verify_session(reorder_log, sid, pk, require_signatures=True)
+    print("reorder verify:", status)
+    assert_false(ok, "Reordered events must FAIL")
+
+    # --- FAIL: delete a line ---
+    deleted = lines[:]
+    deleted.pop(2)
+    del_log = os.path.join(td, "tamper_delete.jsonl")
+    write_lines(del_log, deleted)
+
+    status, ok, report = fr.verify_session(del_log, sid, pk, require_signatures=True)
+    print("delete verify:", status)
+    assert_false(ok, "Deleted event must FAIL")
+
+    # --- FAIL: wrong public key ---
+    sk2, pk2 = fr.gen_keys()
+    status, ok, report = fr.verify_session(log, sid, pk2, require_signatures=True)
+    print("wrong pk verify:", status)
+    assert_false(ok, "Wrong public key must FAIL")
+
+    # --- FAIL: require signatures but unsigned event exists ---
+    # create new session with mixed signed/unsigned
+    sid2, _ = fr.start_session(log, "audit-demo", "deterministic", "mixed sigs", False, sk)
+    ev, _ = fr.append_event(log, sid2, "note", '{"x":1}', "", False, sk, "audit-demo", "deterministic")  # unsigned
+    status, ok, report = fr.verify_session(log, sid2, pk, require_signatures=True)
+    print("require sigs w/ unsigned:", status)
+    assert_false(ok, "Require signatures must fail if any event unsigned")
+
+    print("\nALL BRUTAL TESTS PASSED (meaning PASS/FAIL behaved correctly).")
+    shutil.rmtree(td, ignore_errors=True)
+    # leave bundle in cwd for inspection
+    print("Bundle left in cwd:", bundle)
+
+if __name__ == "__main__":
+    main()