chore: initial commit — recon artifacts + design spec + Phase 1 plan

recon/scripts/analyze_flows.py

@@ -0,0 +1,46 @@
+"""Read mitmproxy flow file, print unique endpoints by host/method/path."""
+import sys, collections
+from mitmproxy import io as miio
+from mitmproxy.exceptions import FlowReadException
+
+flow_file = sys.argv[1]
+filter_host = sys.argv[2] if len(sys.argv) > 2 else None
+
+stats = collections.Counter()
+endpoints = collections.defaultdict(set)
+hosts = collections.Counter()
+status_codes = collections.Counter()
+
+with open(flow_file, "rb") as f:
+    reader = miio.FlowReader(f)
+    try:
+        for flow in reader.stream():
+            if not hasattr(flow, "request"):
+                continue
+            req = flow.request
+            host = req.pretty_host
+            if filter_host and filter_host not in host:
+                continue
+            hosts[host] += 1
+            key = (req.method, req.path.split("?")[0])
+            endpoints[host].add(key)
+            stats[(req.method, host, req.path.split("?")[0])] += 1
+            if flow.response:
+                status_codes[flow.response.status_code] += 1
+    except FlowReadException as e:
+        print(f"[warn] flow truncated: {e}", file=sys.stderr)
+
+print("=== Hosts ===")
+for h, c in hosts.most_common():
+    print(f"  {c:5d}  {h}")
+print()
+print("=== Endpoints grouped by host ===")
+for h in sorted(endpoints.keys()):
+    print(f"\n--- {h} ---")
+    for method, path in sorted(endpoints[h]):
+        count = stats[(method, h, path)]
+        print(f"  [{count:3d}x] {method:6s} {path}")
+print()
+print("=== Status codes ===")
+for code, c in status_codes.most_common():
+    print(f"  {code}: {c}")

recon/scripts/dump_response.py

@@ -0,0 +1,43 @@
+"""Dump request URL + response body for flows matching a filter substring."""
+import sys, json
+from mitmproxy import io as miio
+from mitmproxy.exceptions import FlowReadException
+
+flow_file = sys.argv[1]
+filter_substr = sys.argv[2]
+max_dumps = int(sys.argv[3]) if len(sys.argv) > 3 else 1
+
+dumps = 0
+with open(flow_file, "rb") as f:
+    reader = miio.FlowReader(f)
+    try:
+        for flow in reader.stream():
+            if not hasattr(flow, "request"):
+                continue
+            url = flow.request.pretty_url
+            if filter_substr not in url:
+                continue
+            if not flow.response:
+                continue
+            dumps += 1
+            print(f"\n=== {flow.request.method} {url} ===")
+            print(f"Status: {flow.response.status_code}")
+            print(f"Content-Type: {flow.response.headers.get('Content-Type', '?')}")
+            print("--- response (first 4000 chars) ---")
+            try:
+                body = flow.response.get_text(strict=False)
+                if body:
+                    try:
+                        parsed = json.loads(body)
+                        body = json.dumps(parsed, indent=2, ensure_ascii=False)
+                    except (json.JSONDecodeError, ValueError):
+                        pass
+                    print(body[:4000])
+                else:
+                    print("(binary or empty)")
+            except Exception as e:
+                print(f"[error decoding body: {e}]")
+            if dumps >= max_dumps:
+                break
+    except FlowReadException as e:
+        print(f"[warn] flow truncated: {e}", file=sys.stderr)

recon/scripts/extract_plain_pe.py

@@ -0,0 +1,47 @@
+"""Scan a binary blob for MZ markers, validate as PE, dump each as separate file."""
+import struct, sys, pathlib
+
+blob_path = pathlib.Path(sys.argv[1])
+out_dir = pathlib.Path(sys.argv[2])
+out_dir.mkdir(parents=True, exist_ok=True)
+
+data = blob_path.read_bytes()
+
+def pe_total_size(buf, off):
+    """Compute size of PE image at offset using section table."""
+    # MZ at off; PE header offset at off+0x3C
+    if off + 0x40 > len(buf):
+        return None
+    pe_off = struct.unpack_from("<I", buf, off + 0x3C)[0]
+    pe_start = off + pe_off
+    if pe_start + 24 > len(buf) or buf[pe_start:pe_start+4] != b"PE\x00\x00":
+        return None
+    num_sections = struct.unpack_from("<H", buf, pe_start + 6)[0]
+    optional_header_size = struct.unpack_from("<H", buf, pe_start + 20)[0]
+    sections_start = pe_start + 24 + optional_header_size
+    last_end = sections_start + num_sections * 40
+    for i in range(num_sections):
+        s = sections_start + i * 40
+        raw_size = struct.unpack_from("<I", buf, s + 16)[0]
+        raw_off = struct.unpack_from("<I", buf, s + 20)[0]
+        end = off + raw_off + raw_size
+        last_end = max(last_end, end)
+    return last_end - off
+
+i = 0
+n = 0
+while True:
+    j = data.find(b"MZ", i)
+    if j == -1:
+        break
+    sz = pe_total_size(data, j)
+    if sz and 4096 < sz < 5_000_000:
+        out = out_dir / f"plain_{j:08x}.dll"
+        out.write_bytes(data[j:j+sz])
+        print(f"[ok] offset=0x{j:08x} size={sz} -> {out.name}")
+        n += 1
+        i = j + sz
+    else:
+        i = j + 2
+
+print(f"\nExtracted {n} plain PE files into {out_dir}")

recon/scripts/grep_utf16.py

@@ -0,0 +1,22 @@
+"""Search a binary as UTF-16 LE, return regex matches."""
+import re, sys, pathlib, glob
+
+pattern = sys.argv[1]
+paths = sys.argv[2:]
+rx = re.compile(pattern)
+
+found = {}
+for pat in paths:
+    for p in glob.glob(pat):
+        try:
+            data = pathlib.Path(p).read_bytes().decode("utf-16-le", errors="ignore")
+        except Exception as e:
+            print(f"[skip] {p}: {e}", file=sys.stderr)
+            continue
+        for m in rx.findall(data):
+            key = m if isinstance(m, str) else m[0]
+            found.setdefault(key, []).append(p)
+
+for key in sorted(found.keys()):
+    files = sorted(set(pathlib.Path(f).name for f in found[key]))
+    print(f"{','.join(files):20s}  {key}")

recon/scripts/unpack_xaba.py

@@ -0,0 +1,49 @@
+"""
+Distill XABA/XALZ assemblies.blob into individual .NET assemblies.
+Format ref (reverse-engineered): https://github.com/dotnet/android/blob/main/src/Xamarin.Android.Build.Tasks/Utilities/AssemblyStore.cs
+"""
+import struct, sys, pathlib, lz4.block
+
+if len(sys.argv) != 3:
+    print("Usage: unpack_xaba.py <assemblies.blob> <out_dir>")
+    sys.exit(1)
+
+blob_path = pathlib.Path(sys.argv[1])
+out_dir = pathlib.Path(sys.argv[2])
+out_dir.mkdir(parents=True, exist_ok=True)
+
+data = blob_path.read_bytes()
+assert data[:4] == b"XABA", f"not a XABA blob, got {data[:4]!r}"
+
+# Scan for XALZ markers (each = one compressed assembly)
+i = 0
+extracted = 0
+while True:
+    idx = data.find(b"XALZ", i)
+    if idx == -1:
+        break
+    # XALZ: 4 magic + 4 descriptor_index + 4 uncompressed_size + N lz4_data
+    desc_idx = struct.unpack_from("<I", data, idx + 4)[0]
+    uncompressed = struct.unpack_from("<I", data, idx + 8)[0]
+    payload_start = idx + 12
+    # Find next XALZ to bound payload (or EOF)
+    next_idx = data.find(b"XALZ", payload_start)
+    payload_end = next_idx if next_idx != -1 else len(data)
+    compressed = data[payload_start:payload_end]
+    try:
+        decompressed = lz4.block.decompress(compressed, uncompressed_size=uncompressed)
+    except Exception as e:
+        # Try without explicit size as fallback
+        try:
+            decompressed = lz4.block.decompress(compressed)
+        except Exception as e2:
+            print(f"[skip] desc={desc_idx} decompress failed: {e2}")
+            i = payload_start
+            continue
+    out_file = out_dir / f"{desc_idx:03d}.dll"
+    out_file.write_bytes(decompressed)
+    extracted += 1
+    print(f"[ok]   desc={desc_idx:03d}  size={len(decompressed)}  -> {out_file.name}")
+    i = payload_start
+
+print(f"\nExtracted {extracted} assemblies into {out_dir}")