Joseph Doherty
fe2a6db786
rust / build / test / clippy / fmt (push) Has been cancelled
Layout:
- src/ .NET 10 x64 reference: MxNativeCodec, MxNativeClient,
MxAsbClient, probes, tests, harnesses. Executable spec.
- design/ Architectural plan for the Rust port (M0–M6), error
model, protocol invariants, risks (R1–R16), adversarial
review log (review.md).
- rust/ Rust workspace. M0 skeleton + M1 codec parity.
mxaccess-codec: 215 unit tests + 2 cross-implementation
parity tests (byte-identical against .NET reference).
Other crates are M0 stubs awaiting M2+.
- captures/ Frida + netsh + pcap evidence per CLAUDE.md
("captures are evidence, not throwaway logs").
- analysis/ Decompiled C# (frida/proxy/decompiled-*),
Ghidra exports for native DLLs (`exports/` only —
working state at `projects/` and AVEVA's input
binaries at `input/` are gitignored).
- docs/ Reverse-engineering reference docs.
- tools/ Setup-LiveProbeEnv.ps1 (Infisical credential fetcher),
Compute-Crc.ps1 (.NET parity helper).
- .github/workflows/ Rust CI: fmt + build + test + clippy on Windows.
- LICENSE MIT (Joseph Doherty, 2026).
Verified:
- cargo test --workspace → 217 passed (215 unit + 2 .NET parity), 0 failed
- cargo clippy --workspace -- -D warnings → clean
- cargo fmt --all -- --check → clean
- cargo publish --dry-run -p mxaccess-codec → packages cleanly
Excluded from history (see .gitignore):
- **/bin, **/obj, **/target — build artifacts
- analysis/ghidra/projects/ — Ghidra working state (regenerable)
- analysis/ghidra/input/ — AVEVA proprietary DLLs (vendor IP)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
|
|
HEADER = [
|
|
"capture",
|
|
"stream",
|
|
"packet_type",
|
|
"context_id",
|
|
"opnum",
|
|
"count",
|
|
"frag_lengths",
|
|
]
|
|
|
|
|
|
def summarize(path: Path) -> list[list[str]]:
|
|
rows: list[list[str]] = []
|
|
counts: Counter[tuple[str, str, str, str]] = Counter()
|
|
lengths: dict[tuple[str, str, str, str], Counter[str]] = {}
|
|
|
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
|
reader = csv.reader(handle, delimiter="\t")
|
|
for fields in reader:
|
|
if len(fields) < 10:
|
|
continue
|
|
|
|
stream = fields[2]
|
|
packet_type = fields[3]
|
|
context_id = fields[5]
|
|
opnum = fields[6]
|
|
frag_len = fields[8]
|
|
key = (stream, packet_type, context_id, opnum)
|
|
counts[key] += 1
|
|
lengths.setdefault(key, Counter())[frag_len] += 1
|
|
|
|
for key, count in sorted(counts.items(), key=lambda item: (-item[1], item[0])):
|
|
stream, packet_type, context_id, opnum = key
|
|
frag_lengths = ",".join(
|
|
f"{length}:{length_count}"
|
|
for length, length_count in sorted(lengths[key].items(), key=lambda item: (item[0], item[1]))
|
|
)
|
|
rows.append([path.parent.name, stream, packet_type, context_id, opnum, str(count), frag_lengths])
|
|
|
|
return rows
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("dcerpc_tsv", type=Path, nargs="+")
|
|
parser.add_argument("--out", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
output_rows = [HEADER]
|
|
for path in args.dcerpc_tsv:
|
|
output_rows.extend(summarize(path))
|
|
|
|
if args.out:
|
|
args.out.parent.mkdir(parents=True, exist_ok=True)
|
|
with args.out.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.writer(handle, delimiter="\t", lineterminator="\n")
|
|
writer.writerows(output_rows)
|
|
else:
|
|
writer = csv.writer(__import__("sys").stdout, delimiter="\t", lineterminator="\n")
|
|
writer.writerows(output_rows)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|