goober/nte-exporter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

0.1.3 - Add user UID, capture source & TCP support

Browse source 
Introduce automatic user UID extraction and optional --user-uid override, include user_uid and capture_source in exported JSON, and prepend UID to export filenames (falls back to unknown). Propagate detected protocol (tcp/udp) through live/libpcap parsing and accept TCP packets for UID detection; adjust libpcap filter to capture TCP as well. Update CLI flags (--copy-clipboard semantics, --user-uid), prompt user when UID is missing, and wire UID through mitmproxy and live capture flows. Bump package/exporter version to 0.1.3 and add tests for UID extraction and filename behavior.
This commit is contained in:
Golumpa 2026-06-17 11:20:31 +01:00
parent 98d97cb36a
commit df1add6cf1
15 changed files with 283 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

20
README.md
View file

@ -25,7 +25,7 @@ Prototype CLI exporter for **Neverness to Everness** pull history — decodes yo
The exporter decodes Permanent Board, Limited Character Board, and Arc Miracle Box history pages from captured UDP data, applies conservative timestamp-boundary handling, and writes sanitized JSON suitable for tracker import.
> [!NOTE]
> The import JSON contains decoded history rows only. It does **not** export tokens, account IDs, role IDs, device IDs, server IPs, raw packets, cookies, session data, or any other capture metadata.
> The import JSON contains decoded history rows and the shareable NTE user UID when it can be detected. It does **not** export tokens, account IDs, role IDs, device IDs, server IPs, raw packets, cookies, session data, or other capture metadata.
## Requirements
@ -57,15 +57,17 @@ Use `--capture-backend libpcap` to require Npcap/libpcap without fallback, or `-
Or simply double-click **`run-exporter.cmd`** — it asks for confirmation before requesting Administrator privileges, and does nothing until you agree.
> [!IMPORTANT]
> Launch the tool **before pressing Start on the game's main menu** so the game's UDP connection can be captured. If you are already in game, log out to the main menu and enter again.
> For automatic user UID detection, launch the tool **before pressing Start on the game's main menu**. If you are already in game, history capture can still work; the tool will ask for your UID before saving if it cannot detect it automatically.
Once running, open any supported history board in game. The tool keeps listening until you press any key. Exports are written under `exports\` as:
- `Permanent_<date_time>.json`
- `Limited_<date_time>.json`
- `Arc_<date_time>.json`
- `<user_uid>_Permanent_<date_time>.json`
- `<user_uid>_Limited_<date_time>.json`
- `<user_uid>_Arc_<date_time>.json`
If only one banner is captured, the export JSON is copied to your clipboard. If multiple banners are captured in the same run, clipboard copy is skipped so one banner does not overwrite another.
If the user UID is not detected automatically, the console asks for it before saving. Leaving it blank saves as `unknown_<banner>_<date_time>.json`, but may prevent import on some trackers.
Exports are not copied to the clipboard by default. Add `--copy-clipboard` to copy a single captured banner's JSON after saving. If multiple banners are captured in the same run, clipboard copy is skipped so one banner does not overwrite another.
If a page response is missed, the exporter reports the missing page number while capture is still running. Leave the exporter open, close and reopen that history board, then scroll down again. Scrolling backward within the existing view does not request the cached pages again. The replacement capture is accepted and the tool confirms when the gap has been recovered. If reopening the board still produces no page messages, return to the main menu and re-enter the game to start a fresh connection.
@ -96,6 +98,8 @@ Decodes a `mitmproxy .flows` capture instead of listening live — used for rese
| --------- | --------------------------------------------------- |
| `--live` | Capture live UDP traffic instead of reading a file. |
| `--debug` | Also write the full research CSV next to each JSON. |
| `--user-uid <uid>` | Override the auto-detected NTE user UID in the JSON export. |
| `--copy-clipboard` | Copy a single live export JSON to clipboard after saving. |
Advanced live-capture selection:
@ -107,13 +111,15 @@ Advanced live-capture selection:
The `--debug` CSV holds any extra information that might be needed for fixing bugs. It contains no dangerous personal account data — only the raw bytes of the captured history page.
The exporter automatically includes the shareable NTE user UID when it appears in the capture. If a short capture does not include it, the console asks before saving; you can also pass it explicitly with `--user-uid`.
> [!TIP]
> For reliable deduplication, start from page 1 and scroll through the pages. If you only want pages 1–5, scroll through to page 6 as well just to be on the safe side.
## Privacy
> [!CAUTION]
> Do not commit packet captures, generated exports, research briefs, or personal account data. The repository keeps `exports/` as an empty output folder but ignores everything generated inside it.
> Sanitized exports are intended for tracker import and should not contain anything especially harmful, but they can identify the game account via user UID and pull history. Share exports only with verified sources, such as known trackers. Do not commit packet captures, generated exports, research briefs, or personal account data. The repository keeps `exports/` as an empty output folder but ignores everything generated inside it.
## Boundary Policy

7
docs/export-format.md
View file

@ -8,9 +8,10 @@ The sanitized JSON export uses:
"format_version": 1,
"game": "Neverness to Everness",
"source": "packet_capture",
"capture_source": "npcap",
"exporter": {
"name": "nte-history-exporter",
"version": "0.1.1"
"version": "0.1.3"
},
"banner": {
"id": "Lottery_Permanent",
@ -26,6 +27,7 @@ The sanitized JSON export uses:
"skipped_records": 0,
"warnings": []
},
"user_uid": "optional-user-uid",
"records": []
}
```
@ -85,3 +87,6 @@ Example Arc record:
```
Normal JSON exports do not include raw packets or capture-only metadata.
`user_uid` is included when detected automatically or supplied with `--user-uid`.
`capture_source` records the capture backend/source used for the export, such as
`npcap`, `libpcap`, `windows_packet`, or `mitmproxy_flows`.

2
pyproject.toml
View file

@ -1,6 +1,6 @@
[project]
name = "nte-history-exporter"
version = "0.1.2"
version = "0.1.3"
description = "Cross-platform Neverness to Everness pull-history exporter."
requires-python = ">=3.10"
dependencies = []

2
src/nte_history_exporter/__init__.py
View file

@ -1 +1 @@
__version__ = "0.1.2"
__version__ = "0.1.3"

9
src/nte_history_exporter/adapters/mitmproxy_flows.py
View file

@ -1,11 +1,13 @@
from __future__ import annotations
from collections import Counter
from pathlib import Path
from typing import Any
from nte_history_exporter.decoder.boundary import select_continuous_run_from_page_1
from nte_history_exporter.decoder.arc import build_arc_rows_from_pairs, select_continuous_arc_run
from nte_history_exporter.decoder.run import build_rows_from_pairs
from nte_history_exporter.decoder.user_uid import extract_user_uid_candidates
from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket
@ -77,6 +79,12 @@ def find_udp_flow(flows: list[Any], preferred_index: int | None = None) -> tuple
def decode_mitmproxy_flows(path: str | Path, flow_index: int | None = None) -> dict[str, Any]:
flows = read_flows(path)
user_uid_candidates: Counter[str] = Counter()
for flow in flows:
for msg in flow.get(b"messages", []):
user_uid_candidates.update(extract_user_uid_candidates(msg[1]))
user_uid = user_uid_candidates.most_common(1)[0][0] if user_uid_candidates else None
resolved_flow_index, flow = find_udp_flow(flows, flow_index)
messages = flow[b"messages"]
@ -110,4 +118,5 @@ def decode_mitmproxy_flows(path: str | Path, flow_index: int | None = None) -> d
"best_arc_run": best_arc_run,
"arc_rows": arc_rows,
"arc_warnings": arc_warnings,
"user_uid": session.user_uid or user_uid,
}

14
src/nte_history_exporter/cli.py
View file

@ -26,8 +26,9 @@ def build_parser() -> argparse.ArgumentParser:
default="auto",
help="live capture backend; auto prefers Npcap/libpcap and falls back to raw sockets on Windows",
)
parser.add_argument("--no-clipboard", action="store_true", help="do not copy live exports to clipboard")
parser.add_argument("--copy-clipboard", action="store_true", help="copy a single live export to clipboard")
parser.add_argument("--debug", action="store_true", help="also write research CSVs next to the JSON exports")
parser.add_argument("--user-uid", default=None, help="override the auto-detected NTE user UID in the JSON export")
return parser
@ -39,8 +40,9 @@ def main(argv: list[str] | None = None) -> int:
run_live_capture(
interface_ip=args.interface_ip,
capture_backend=args.capture_backend,
copy_clipboard=not args.no_clipboard,
copy_clipboard=args.copy_clipboard,
write_debug_csv=args.debug,
user_uid=args.user_uid,
)
return 0
except (LibpcapUnavailable, PermissionError) as exc:
@ -62,13 +64,19 @@ def main(argv: list[str] | None = None) -> int:
best_run = decoded["best_run"]
pair_count = len(decoded["pairs"])
out_path, json_path = export_paths(kind)
resolved_user_uid = args.user_uid or decoded.get("user_uid")
if not resolved_user_uid:
resolved_user_uid = console.prompt_user_uid()
out_path, json_path = export_paths(kind, resolved_user_uid)
if args.debug:
write_csv(out_path, rows)
export = build_export_json(
rows,
warnings,
source="packet_capture",
capture_source="mitmproxy_flows",
user_uid=resolved_user_uid,
flow_index=decoded["flow_index"],
candidate_request_response_pairs=pair_count,
pages_seen=[p[0] for p in best_run],

17
src/nte_history_exporter/console.py
View file

@ -68,10 +68,10 @@ def print_live_instructions(local_ip: str, backend: str = "windows_raw", detail:
print(style(f" Capture backend: {backend}{backend_detail}", DIM))
print()
print(style(" How to export your pull history", BOLD))
print(" 1. This tool must be running BEFORE you press Start on")
print(" the game's main menu, or the game connection cannot")
print(" be captured. Already in game? Log out to the main")
print(" menu and enter again.")
print(" 1. For automatic user UID detection, start this tool")
print(" before pressing Start on the game's main menu.")
print(" Already in game? You can still capture history;")
print(" the tool will ask for your UID if it cannot detect it.")
print(" 2. Open a supported history screen:")
print(style(" Monopoly > Standard Board history", CYAN))
print(style(" Monopoly > Limited Character Board history", CYAN))
@ -151,3 +151,12 @@ def print_success(text: str) -> None:
def print_problem(text: str) -> None:
print(style(f" {text}", YELLOW, BOLD))
def prompt_user_uid() -> str | None:
print()
print_problem("User UID was not detected in this capture.")
print_note("Enter your NTE user UID so the export can be named and linked correctly.")
print_note("Leaving this blank may prevent import on some trackers.")
value = input(" User UID: ").strip()
return value or None

2
src/nte_history_exporter/constants.py
View file

@ -11,7 +11,7 @@ LIMITED_CHARACTER_BANNER_NAME = "Limited Character Board"
ARC_BANNER_ID = "Arc_MiracleBox"
ARC_BANNER_NAME = "Arc Miracle Box"
EXPORTER_NAME = "nte-history-exporter"
EXPORTER_VERSION = "0.1.2"
EXPORTER_VERSION = "0.1.3"
HISTORY_REQUEST_BANNER = 4220
HISTORY_REQUEST_LENGTH = 45

46
src/nte_history_exporter/decoder/user_uid.py Normal file
View file

@ -0,0 +1,46 @@
from __future__ import annotations
import struct
from collections import Counter
MIN_USER_UID = 100_000_000_000
MAX_USER_UID = 999_999_999_999
USER_UID_RECORD_OFFSETS = (
(b"TagOthers", (16,)),
(b"PrivateSpawnInfoRecord", (40,)),
(b"SimpleQuestRecord", (20, 52)),
(b"FurnitureLayoutDataRec", (56, 64)),
(b"StoreBrandItemSalesVolumeRecord", (16,)),
(b"StorePropertyRecord", (20,)),
)
def _plausible_user_uid(value: int) -> bool:
return MIN_USER_UID <= value <= MAX_USER_UID
def extract_user_uid_candidates(payload: bytes) -> list[str]:
results = []
for marker, distances in USER_UID_RECORD_OFFSETS:
search_from = 0
while True:
marker_pos = payload.find(marker, search_from)
if marker_pos == -1:
break
for distance in distances:
offset = marker_pos - distance
if offset < 0:
continue
value = struct.unpack_from("<Q", payload, offset)[0]
if _plausible_user_uid(value):
results.append(str(value))
search_from = marker_pos + len(marker)
return results
def extract_user_uid(payload: bytes) -> str | None:
candidates = extract_user_uid_candidates(payload)
if not candidates:
return None
return Counter(candidates).most_common(1)[0][0]

32
src/nte_history_exporter/export/json_export.py
View file

@ -17,6 +17,8 @@ def build_export_json(
warnings: list[dict[str, Any]],
*,
source: str = "packet_capture",
capture_source: str | None = None,
user_uid: str | None = None,
flow_index: int | None = None,
candidate_request_response_pairs: int | None = None,
pages_seen: list[int] | None = None,
@ -40,21 +42,31 @@ def build_export_json(
if pages_seen is not None:
scan["pages_seen"] = pages_seen
return {
normalized_user_uid = user_uid.strip() if user_uid else ""
export: dict[str, Any] = {
"format": "nte-history-export",
"format_version": 1,
"game": GAME_NAME,
"source": source,
"exporter": {"name": EXPORTER_NAME, "version": __version__},
"banner": {
"id": pool["id"],
"name": pool["name"],
"system": pool["system"],
"shared_pity": pool["shared_pity"],
},
"scan": scan,
"records": [_record_for_export(r) for r in exported],
}
if capture_source:
export["capture_source"] = capture_source
export["exporter"] = {"name": EXPORTER_NAME, "version": __version__}
export.update(
{
"banner": {
"id": pool["id"],
"name": pool["name"],
"system": pool["system"],
"shared_pity": pool["shared_pity"],
},
"scan": scan,
}
)
if normalized_user_uid:
export["user_uid"] = normalized_user_uid
export["records"] = [_record_for_export(r) for r in exported]
return export
def _record_for_export(row: dict[str, Any]) -> dict[str, Any]:

6
src/nte_history_exporter/live_capture/libpcap.py
View file

@ -10,7 +10,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
from nte_history_exporter.live_capture.windows_raw import ParsedIpUdpPacket, parse_ipv4_udp_packet
from nte_history_exporter.live_capture.windows_raw import ParsedIpUdpPacket, parse_ipv4_packet
PCAP_ERRBUF_SIZE = 256
SNAP_LENGTH = 65535
@ -351,7 +351,7 @@ class LibpcapCapture:
raise LibpcapUnavailable(_pcap_error(self.lib, self.handle))
program = _BpfProgram()
filter_expression = f"udp and host {local_ip}".encode("ascii")
filter_expression = f"host {local_ip} and (udp or tcp)".encode("ascii")
if self.lib.pcap_compile(self.handle, ctypes.byref(program), filter_expression, 1, 0xFFFFFFFF) != 0:
raise LibpcapUnavailable(_pcap_error(self.lib, self.handle))
try:
@ -382,7 +382,7 @@ class LibpcapCapture:
ipv4_packet = _extract_ipv4_frame(frame, self.datalink)
if ipv4_packet is None:
continue
packet = parse_ipv4_udp_packet(ipv4_packet)
packet = parse_ipv4_packet(ipv4_packet)
if packet is not None:
yield packet

39
src/nte_history_exporter/live_capture/runner.py
View file

@ -24,6 +24,10 @@ EXPORT_PREFIXES = {
"arc_miracle_box": "Arc",
}
CAPTURE_SOURCE_LABELS = {
"windows_raw": "windows_packet",
}
def copy_to_clipboard(text: str) -> bool:
try:
@ -60,8 +64,9 @@ def run_live_capture(
*,
interface_ip: str | None = None,
capture_backend: str = "auto",
copy_clipboard: bool = True,
copy_clipboard: bool = False,
write_debug_csv: bool = False,
user_uid: str | None = None,
) -> dict:
local_ip = interface_ip or detect_local_ipv4()
session = LiveHistorySession(local_ip)
@ -88,6 +93,7 @@ def run_live_capture(
src_port=packet.src_port,
dst_port=packet.dst_port,
payload=packet.payload,
protocol=packet.protocol,
)
)
if matched:
@ -130,6 +136,10 @@ def run_live_capture(
)
exports = []
resolved_user_uid = user_uid or session.user_uid
if session.kinds_seen() and not resolved_user_uid:
resolved_user_uid = console.prompt_user_uid()
capture_source = CAPTURE_SOURCE_LABELS.get(capture.name, capture.name)
for kind in session.kinds_seen():
pairs = session.pairs_for_kind(kind)
best_run, run_warnings = select_continuous_run_from_page_1(pairs)
@ -138,13 +148,15 @@ def run_live_capture(
rows = annotate_groups(rows)
warnings = run_warnings
pages_seen = [p[0] for p in best_run]
csv_path, json_path = export_paths(kind)
csv_path, json_path = export_paths(kind, resolved_user_uid)
if write_debug_csv:
write_csv(csv_path, rows)
export = build_export_json(
rows,
warnings,
source="live_capture",
capture_source=capture_source,
user_uid=resolved_user_uid,
pages_seen=pages_seen,
)
payload = json.dumps(export, ensure_ascii=False, indent=2)
@ -162,9 +174,9 @@ def run_live_capture(
console.print_results_header()
if not exports:
console.print_problem("No history pages were captured.")
console.print_note("This tool must already be running when you press Start on the")
console.print_note("game's main menu. Log out to the main menu, enter the game")
console.print_note("again, then reopen the history screen.")
console.print_note("Make sure the capture backend is running, then reopen the")
console.print_note("history screen and scroll from page 1. If no page messages")
console.print_note("appear, return to the main menu and re-enter the game.")
return {"exports": []}
for item in exports:
@ -189,24 +201,33 @@ def run_live_capture(
console.print_success("Export copied to clipboard - paste it straight into your tracker.")
else:
console.print_note("Clipboard tool unavailable; use the JSON file shown above.")
elif len(exports) > 1:
elif copy_clipboard and len(exports) > 1:
console.print_note("Multiple banners captured; clipboard copy skipped so one export")
console.print_note("does not overwrite another.")
return {"exports": exports}
def export_paths(kind: str) -> tuple[Path, Path]:
def export_paths(kind: str, user_uid: str | None = None) -> tuple[Path, Path]:
export_dir = Path("exports")
export_dir.mkdir(parents=True, exist_ok=True)
prefix = EXPORT_PREFIXES.get(kind, "History")
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
base = export_dir / f"{prefix}_{stamp}"
uid_prefix = _safe_filename_part(user_uid) if user_uid else "unknown"
base_name = f"{uid_prefix}_{prefix}_{stamp}"
base = export_dir / base_name
csv_path = base.with_suffix(".csv")
json_path = base.with_suffix(".json")
counter = 2
while csv_path.exists() or json_path.exists():
base = export_dir / f"{prefix}_{stamp}_{counter}"
base = export_dir / f"{base_name}_{counter}"
csv_path = base.with_suffix(".csv")
json_path = base.with_suffix(".json")
counter += 1
return csv_path, json_path
def _safe_filename_part(value: str | None) -> str:
if not value:
return "unknown"
cleaned = "".join(ch for ch in value.strip() if ch.isalnum() or ch in ("-", "_"))
return cleaned or "unknown"

12
src/nte_history_exporter/live_capture/session.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from collections import deque
from collections import Counter, deque
from dataclasses import dataclass
from typing import Any
@ -19,6 +19,7 @@ from nte_history_exporter.decoder.protocol import (
request_page,
)
from nte_history_exporter.decoder.run import build_rows_from_pairs
from nte_history_exporter.decoder.user_uid import extract_user_uid_candidates
@dataclass
@ -29,6 +30,7 @@ class UdpPacket:
src_port: int
dst_port: int
payload: bytes
protocol: str = "udp"
@dataclass
@ -57,6 +59,8 @@ class LiveHistorySession:
self.last_capture_was_replacement = False
self.requested_pages: dict[str, set[int]] = {}
self.unanswered_pages: dict[str, dict[int, str]] = {}
self.user_uid: str | None = None
self.user_uid_candidates: Counter[str] = Counter()
def _mark_unanswered(self, request: PendingRequest) -> None:
if request.response_candidates:
@ -93,6 +97,12 @@ class LiveHistorySession:
def process_packet(self, packet: UdpPacket) -> bool:
self.packet_count += 1
candidates = extract_user_uid_candidates(packet.payload)
if candidates:
self.user_uid_candidates.update(candidates)
self.user_uid = self.user_uid_candidates.most_common(1)[0][0]
if packet.protocol != "udp":
return False
if packet.src_ip == self.local_ip and is_history_request(packet.payload):
offset = int.from_bytes(packet.payload[31:35], "little")

37
src/nte_history_exporter/live_capture/windows_raw.py
View file

@ -10,12 +10,16 @@ RECEIVE_BUFFER_SIZE = 4 * 1024 * 1024
@dataclass
class ParsedIpUdpPacket:
class ParsedIpPacket:
src_ip: str
dst_ip: str
src_port: int
dst_port: int
payload: bytes
protocol: str = "udp"
ParsedIpUdpPacket = ParsedIpPacket
def detect_local_ipv4() -> str:
@ -42,24 +46,41 @@ def detect_local_ipv4() -> str:
return candidates[0]
def parse_ipv4_udp_packet(data: bytes) -> ParsedIpUdpPacket | None:
def parse_ipv4_packet(data: bytes) -> ParsedIpPacket | None:
if len(data) < 28:
return None
version_ihl = data[0]
if version_ihl >> 4 != 4:
return None
ihl = (version_ihl & 0x0F) * 4
total_length = struct.unpack_from("!H", data, 2)[0]
if total_length > 0:
data = data[:total_length]
if len(data) < ihl + 8:
return None
protocol = data[9]
if protocol != 17:
return None
src_ip = socket.inet_ntoa(data[12:16])
dst_ip = socket.inet_ntoa(data[16:20])
src_port, dst_port, udp_len, _checksum = struct.unpack_from("!HHHH", data, ihl)
payload = data[ihl + 8 : ihl + udp_len]
return ParsedIpUdpPacket(src_ip, dst_ip, src_port, dst_port, payload)
if protocol == 17:
src_port, dst_port, udp_len, _checksum = struct.unpack_from("!HHHH", data, ihl)
payload = data[ihl + 8 : ihl + udp_len]
return ParsedIpPacket(src_ip, dst_ip, src_port, dst_port, payload, "udp")
if protocol == 6:
if len(data) < ihl + 20:
return None
src_port, dst_port = struct.unpack_from("!HH", data, ihl)
tcp_header_len = (data[ihl + 12] >> 4) * 4
if tcp_header_len < 20 or len(data) < ihl + tcp_header_len:
return None
payload = data[ihl + tcp_header_len :]
return ParsedIpPacket(src_ip, dst_ip, src_port, dst_port, payload, "tcp")
return None
def parse_ipv4_udp_packet(data: bytes) -> ParsedIpPacket | None:
packet = parse_ipv4_packet(data)
return packet if packet and packet.protocol == "udp" else None
def open_raw_udp_socket(local_ip: str) -> socket.socket:
@ -79,7 +100,7 @@ def read_packets(sock: socket.socket):
except socket.timeout:
yield None
continue
packet = parse_ipv4_udp_packet(data)
packet = parse_ipv4_packet(data)
if packet is None:
continue
yield packet

87
tests/test_boundary_export.py
View file

@ -20,6 +20,7 @@ from nte_history_exporter.decoder.protocol import decode_response_records, histo
from nte_history_exporter.constants import POOL_META
from nte_history_exporter.mappings import ARC_META, CHARACTERS, ITEMS, REWARDS_BY_ID
from nte_history_exporter.decoder.protocol import decode_reward_key, infer_reward_type
from nte_history_exporter.decoder.user_uid import extract_user_uid
from nte_history_exporter.decoder.arc import (
arc_request_page,
build_arc_rows_from_pairs,
@ -39,8 +40,10 @@ from nte_history_exporter.live_capture.libpcap import (
_extract_ipv4_frame,
LibpcapUnavailable,
)
from nte_history_exporter.live_capture.windows_raw import parse_ipv4_packet
from nte_history_exporter.live_capture.backends import open_capture_backend
from nte_history_exporter.export.json_export import build_export_json
from nte_history_exporter.live_capture.runner import export_paths
from nte_history_exporter.pool_mappings import load_pool_mappings, pool_meta_from_mapping
@ -347,10 +350,94 @@ class BoundaryExportTests(unittest.TestCase):
self.assertEqual(export["format"], "nte-history-export")
self.assertIn("exporter", export)
self.assertNotIn("user_uid", export)
self.assertNotIn("record_hex", export["records"][0])
self.assertNotIn("request_msg", export["records"][0])
self.assertNotIn("response_msg", export["records"][0])
def test_export_includes_user_uid_when_provided(self):
rows = load_reference_csv("monopoly_history_poc_13_pages_1_to_5_v4.csv")
annotated = annotate_groups(rows)
export = build_export_json(
annotated,
[],
capture_source="npcap",
user_uid="123456789",
)
self.assertEqual(list(export).index("user_uid"), list(export).index("records") - 1)
self.assertEqual(export["capture_source"], "npcap")
self.assertEqual(export["user_uid"], "123456789")
def test_export_paths_include_user_uid_banner_and_timestamp(self):
_csv_path, json_path = export_paths("limited_character", "218216016349")
self.assertRegex(
json_path.name,
r"^218216016349_Limited_\d{8}_\d{6}(?:_\d+)?\.json$",
)
def test_extracts_user_uid_from_record_context(self):
payload = (
b"\x00" * 24
+ (218216016349).to_bytes(8, "little")
+ b"\x00\x00\x00\x00\x09\x00\x00\x00TagOthers\x00"
)
self.assertEqual(extract_user_uid(payload), "218216016349")
def test_extracts_user_uid_from_private_spawn_record_context(self):
payload = (
b"\x88\x00\x00\x00\x10\x00\x00\x00"
+ (218216016349).to_bytes(8, "little")
+ b"\x08\x00\x0c\x00\x07\x00\x08\x00\x08\x00\x00\x00"
+ b"\x00\x00\x00\x01\x08\x00\x00\x00\x04\x00\x04\x00"
+ b"\x04\x00\x00\x00\x16\x00\x00\x00PrivateSpawnInfoRecord\x00"
)
self.assertEqual(extract_user_uid(payload), "218216016349")
def test_does_not_extract_user_uid_from_wrong_record_offset(self):
payload = (
b"\x00" * 28
+ (218216016349).to_bytes(8, "little")
+ b"\x00\x00\x00\x00TagOthers\x00"
)
self.assertIsNone(extract_user_uid(payload))
def test_does_not_extract_old_eight_digit_false_positive_as_user_uid(self):
payload = (
b"WholeVehicleData\x00\x00\x00\x00\x00o<\x00\x00\x05\x00\x00\x00"
b"\x0b\x00\x00\x00Vehicle015\x00\x0b\x00\x00\x00buyvehicle\x00"
b"\x09\x00\x00\x0015363624\x00\x06\x00\x00\x00"
)
self.assertIsNone(extract_user_uid(payload))
def test_ipv4_parser_extracts_tcp_payload_for_user_uid_detection(self):
payload = (
(218216016349).to_bytes(8, "little")
+ b"\x00\x00\x00\x00\x09\x00\x00\x00TagOthers\x00"
)
tcp_header = bytearray(20)
tcp_header[0:2] = (40000).to_bytes(2, "big")
tcp_header[2:4] = (30000).to_bytes(2, "big")
tcp_header[12] = 5 << 4
total_len = 20 + len(tcp_header) + len(payload)
ip_header = bytearray(20)
ip_header[0] = 0x45
ip_header[2:4] = total_len.to_bytes(2, "big")
ip_header[9] = 6
ip_header[12:16] = bytes([192, 0, 2, 1])
ip_header[16:20] = bytes([198, 51, 100, 2])
packet = parse_ipv4_packet(bytes(ip_header) + bytes(tcp_header) + payload)
self.assertIsNotNone(packet)
self.assertEqual(packet.protocol, "tcp")
self.assertEqual(packet.payload, payload)
def test_limited_selector_and_marker_decode(self):
request = bytearray(45)
request[31:35] = (4).to_bytes(4, "little")