"""Adversarial tests (session requirement 4). Covers: CRLF / LF / mixed endings, exact boundary, boundary inside line, malformed timestamp, malformed price, invalid bid/ask, duplicate data, dropped data, timestamp reversal, worker reorder, incorrect carry, altered prefix, repeated resume, already-processed state. """ import os import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from engine.chunkmap import build_chunk_map # noqa: E402 from engine.config import default_config # noqa: E402 from engine.parse import parse_chunk # noqa: E402 from engine.verify.vparse import v_parse_chunk # noqa: E402 from engine.canonical import tick_chunk_content_id # noqa: E402 from engine.verify.vinvariants import independent_ticks_content_id, \ independent_bars_content_id # noqa: E402 from engine.storage import ( # noqa: E402 ticks_chunk_path, bar_part_path, read_ticks_chunk, read_bar_part, checkpoint_path, cert_path, journal_path, ) from engine.certify import certify_source # noqa: E402 from engine.checkpoint import ( # noqa: E402 load_checkpoint, write_checkpoint, STATUS_COMPLETED, ) from engine.dispatcher import ( # noqa: E402 init_run, IngestRunner, resume_checks, finalize_completed, new_run_id, ) from engine.util import sha256_bytes # noqa: E402 from tests.golden.common import G01_ROWS, case_bytes, case_chunk_nominal, \ golden_cfg, G12_ROWS # noqa: E402 from tests.golden.run_golden import mini_setup, mini_run, _file_ids # noqa: E402 CERT_NOW = 4_102_444_800_000 def _cfg(): return default_config("C:\\u\\s.csv", output_root="C:\\u\\o", source_tz_offset_minutes=0) def run(): cfg = _cfg() res = [] # 1. mixed line endings -> identical canonical rows to pure LF lines = [x.encode() for x in G01_ROWS] mixed = lines[0] + b"\r\n" + lines[1] + b"\n" + lines[2] + b"\r" + lines[3] + b"\n" lf = b"\n".join(lines) + b"\n" recs_m, _mm, _cmm, _fm = parse_chunk(mixed, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) recs_l, _ml, _cl, _fl = parse_chunk(lf, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) id_m = tick_chunk_content_id([(r[0], r[1], r[2], r[4]) for r in recs_m]) id_l = tick_chunk_content_id([(r[0], r[1], r[2], r[4]) for r in recs_l]) res.append(("adversarial_mixed_terminators", len(recs_m) == 4 and id_m == id_l, "")) # 2. exact boundary at several offsets + inside-line boundaries data = lf all_ok = True for nominal in (1, 2, 10, 46, 47, 92, 141): with tempfile.TemporaryDirectory() as td: src = os.path.join(td, "s.txt") with open(src, "wb") as fh: fh.write(data) cm = build_chunk_map(src, nominal, 5_368_709_120) tot = cm["chunks"][0]["byte_start"] == 0 and \ cm["chunks"][-1]["byte_end"] == cm["total_bytes"] for i in range(1, len(cm["chunks"])): tot = tot and cm["chunks"][i]["byte_start"] == \ cm["chunks"][i - 1]["byte_end"] all_ok = all_ok and tot res.append(("adversarial_boundary_matrix", all_ok, "")) # 3. malformed timestamp battery tsv = ["2023.00.01 00:00:00.000", "2023.01.00 00:00:00.000", "2023.02.30 00:00:00.000", "2023.01.01 00:60:00.000", "2023.01.01 00:00:61.000", "2023.01.01 00:00:00.1000", "23.01.01 00:00:00.000", "2023-01-01 00:00:00.000", "2023.01.01T00:00:00.000", "2023.01.01 00:00:00"] n_parse = 0 from engine.util import parse_timestamp_dotted for t in tsv: _v, err = parse_timestamp_dotted(t) if err is not None: n_parse += 1 res.append(("adversarial_malformed_timestamps", n_parse == 8, "%d/10 rejected" % n_parse)) # 4. malformed price battery from engine.util import parse_price_to_micro prices = ["1.23456789", "-0.5", "+0.5", "1e5", "1,000", "", ".5", "5."] n_rej = sum(1 for p in prices if parse_price_to_micro(p, 1_000_000)[1]) res.append(("adversarial_malformed_prices", n_rej == 8, "%d/8 rejected (5. allows 0 frac digits)" % n_rej)) # 5. invalid bid/ask battery ba = [(100000000, 99999999), (1, 1), (0, 100), (100, 0)] csv_row = "2023.01.01 00:00:00.000,%d.%06d,%d.%06d" lines = [csv_row % (b // 1, b % 1000000, a // 1, a % 1000000) for b, a in ba] data_ba = (b"\n".join(x.encode() for x in ["2023.01.01 00:00:00.000,100.000000,100.000100"] + lines) + b"\n") recs, _, ctr, _ = parse_chunk(data_ba, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) rel = ctr["MALFORMED_BID_ASK_RELATION"] + ctr["MALFORMED_PRICE_NONPOSITIVE"] res.append(("adversarial_invalid_bid_ask", rel == 3 and len(recs) == 2, "rel+nonpos=%d canonical=%d" % (rel, len(recs)))) # 6. duplicate data battery dup2 = lf + lf.split(b"\n")[0] + b"\n" recs_d, _, ctr_d, _ = parse_chunk(dup2, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) res.append(("adversarial_duplicate_data", ctr_d["MALFORMED_DUPLICATE"] >= 1 and len(recs_d) == 4, "")) # 7. dropped data: two-row copy must differ from four-row source with tempfile.TemporaryDirectory() as td: src = os.path.join(td, "s.txt") with open(src, "wb") as fh: fh.write(lf) cm = build_chunk_map(src, 25_165_824, 5_368_709_120) full = parse_chunk(lf, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW)[0] short = parse_chunk(b"\n".join(x.encode() for x in G01_ROWS[:2]) + b"\n", chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW)[0] res.append(("adversarial_dropped_data", len(full) == 4 and len(short) == 2, "full=%d short=%d (verifier count check detects)" % (len(full), len(short)))) # 8. timestamp reversal (within chunk) reversal = G12_ROWS # includes a regression data_rev = (b"\n".join(x.encode() for x in reversal) + b"\n") recs_r, _, ctr_r, _ = parse_chunk(data_rev, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) res.append(("adversarial_timestamp_reversal", ctr_r["MALFORMED_NON_MONOTONIC"] == 1 and len(recs_r) == 2, "")) # 9. worker reorder (multi-chunk ids independent of dispatch order) data9 = lf cm9 = build_chunk_map(_tmp_src(data9), 46, 5_368_709_120) def ids_order(order): out = {} for ci in order: ch = cm9["chunks"][ci] recs, *_ = parse_chunk(data9[ch["byte_start"]:ch["byte_end"]], chunk_index=ci, byte_start=ch["byte_start"], global_line_start=0, cfg=cfg, expect_header=(ci == 0), certify_time_ms=CERT_NOW) out["c%d" % ci] = tick_chunk_content_id([(r[0], r[1], r[2], r[4]) for r in recs]) return out oa = [x["index"] for x in cm9["chunks"]] ob = list(reversed(oa)) res.append(("adversarial_worker_reorder", ids_order(oa) == ids_order(ob), "")) # 10. incorrect carry -> FAILED(carry_continuity) with tempfile.TemporaryDirectory() as td: cfgA, certA, cmA, ridA = mini_setup(case_bytes("G01"), td, chunk_bytes=47, workload_bytes=94) mini_run(cfgA, certA, cmA, td, ridA, pause_after=1) pay = load_checkpoint(checkpoint_path(td)) pay["carries"][cfgA["timeframes"][0]]["last_ts_ms"] += 9_000_000_000 write_checkpoint(checkpoint_path(td), pay) _c, reason = resume_checks(cfgA, td, cfgA["source_path"], cmA, certA) if reason is None: rr = IngestRunner(cfgA, td, cfgA["source_path"], cmA["source_id"], cmA, certA, run_id=ridA) st = rr.run() detected = st == "FAILED" else: detected = True res.append(("adversarial_incorrect_carry", detected, "")) # 11. altered prefix (checkpoint source path tampered, no rehash) with tempfile.TemporaryDirectory() as td: cfgA, certA, cmA, ridA = mini_setup(case_bytes("G01"), td, chunk_bytes=47, workload_bytes=94) mini_run(cfgA, certA, cmA, td, ridA, pause_after=1) pay = load_checkpoint(checkpoint_path(td)) pay["source_identity"]["path"] = "C:\\altered\\prefix.txt" with open(checkpoint_path(td), "w", encoding="utf-8") as fh: import json json.dump(pay, fh, sort_keys=True) _c, reason = resume_checks(cfgA, td, cfgA["source_path"], cmA, certA) res.append(("adversarial_altered_prefix", reason is not None, [reason or "not blocked"])) # 12. repeated resume (three resumes -> ids stable) with tempfile.TemporaryDirectory() as td: cfgA, certA, cmA, ridA = mini_setup(case_bytes("G01"), td, chunk_bytes=47, workload_bytes=94) mini_run(cfgA, certA, cmA, td, ridA, pause_after=1) ids_first = None for _ in range(3): _c, reason = resume_checks(cfgA, td, cfgA["source_path"], cmA, certA) if reason is not None: res.append(("adversarial_repeated_resume", False, ["blocked: " + reason])) break rr = IngestRunner(cfgA, td, cfgA["source_path"], cmA["source_id"], cmA, certA, run_id=ridA) rr.run() ids_now = _file_ids(cmA, td, cfgA) if ids_first is None: ids_first = ids_now elif ids_first != ids_now: res.append(("adversarial_repeated_resume", False, ["ids drifted"])) break else: res.append(("adversarial_repeated_resume", True, "")) if not any(r[0] == "adversarial_repeated_resume" for r in res): res.append(("adversarial_repeated_resume", False, ["no outcome"])) # 13. already-processed state: COMPLETED -> further resume blocked with tempfile.TemporaryDirectory() as td: cfgA, certA, cmA, ridA = mini_setup(case_bytes("G01"), td, chunk_bytes=25_165_824, workload_bytes=5_368_709_120) st = mini_run(cfgA, certA, cmA, td, ridA) from engine.evidence import build_evidence build_evidence(td, cmA, cfgA) finalize_completed(cfgA, td, ridA) _c, reason = resume_checks(cfgA, td, cfgA["source_path"], cmA, certA) res.append(("adversarial_already_processed", reason is not None, [reason or "resume allowed after COMPLETED"])) return res def _tmp_src(data): import tempfile fd = tempfile.NamedTemporaryFile(suffix=".txt", delete=False) fd.write(data) fd.close() return fd.name if __name__ == "__main__": res = run() for name, ok, detail in res: print("%-42s %s %s" % (name, "PASS" if ok else "FAIL", detail)) sys.exit(0 if all(r[1] for r in res) else 1)