From cc88955baf68702e7c47ea8b27f7596d637b22a2 Mon Sep 17 00:00:00 2001 From: Golumpa Date: Thu, 16 Jul 2026 01:34:38 +0100 Subject: [PATCH] Added server detection Implemented server detection at the request of NTEWIZ. If not detected, it will ask after collecting pull data. --- README.md | 9 ++- docs/export-format.md | 9 ++- pyproject.toml | 2 +- src/nte_history_exporter/__init__.py | 2 +- .../adapters/mitmproxy_flows.py | 7 ++ src/nte_history_exporter/cli.py | 5 ++ src/nte_history_exporter/console.py | 29 ++++++++ src/nte_history_exporter/constants.py | 2 +- .../decoder/server_region.py | 55 ++++++++++++++ .../export/json_export.py | 8 +++ .../live_capture/runner.py | 4 ++ .../live_capture/session.py | 7 ++ .../live_capture/stop_key.py | 37 +++++++++- tests/support.py | 2 + tests/test_export_contract.py | 71 +++++++++++++++++++ 15 files changed, 242 insertions(+), 7 deletions(-) create mode 100644 src/nte_history_exporter/decoder/server_region.py diff --git a/README.md b/README.md index 7f553b6..48d9e6b 100644 --- a/README.md +++ b/README.md @@ -90,9 +90,9 @@ From source: Or simply double-click **`run-exporter.cmd`**. > [!IMPORTANT] -> 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. +> For automatic user UID and account-server 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 missing account details before saving. -Once running, open any supported history board in game. The tool keeps listening until you press any key. Exports are written under `exports\` as: +Once running, open any supported history board in game. The tool keeps listening until you press any key. After it prints and saves the results, press any key again to close the exporter. Exports are written under `exports\` as: - `_Permanent_.json` - `_Limited_.json` @@ -101,6 +101,11 @@ Once running, open any supported history board in game. The tool keeps listening If the user UID is not detected automatically, the console asks for it before saving. Leaving it blank saves as `unknown__.json`, but may prevent import on some trackers. +The export also includes `server_id` and `account_region` when known. If the +initial TCP server-selection response was missed, the console offers Asia, +America, Europe, and SEA as a numbered choice. The prompt can be left blank to +omit server information. + 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. diff --git a/docs/export-format.md b/docs/export-format.md index b19e11e..1b9ea37 100644 --- a/docs/export-format.md +++ b/docs/export-format.md @@ -15,7 +15,7 @@ decoder-only offsets. "capture_source": "npcap", "exporter": { "name": "nte-history-exporter", - "version": "0.1.11" + "version": "0.1.12" }, "banner": { "id": "Lottery_Permanent", @@ -32,6 +32,8 @@ decoder-only offsets. "warnings": [] }, "user_uid": "optional-user-uid", + "server_id": "23003", + "account_region": "EU", "records": [] } ``` @@ -45,6 +47,11 @@ Top-level fields: - `banner`: The history pool this file belongs to. - `scan`: Export counts and capture warnings. - `user_uid`: Optional game account UID, if known. +- `server_id`: Optional numeric account-server ID, detected from the initial TCP + connection or selected by the user. +- `account_region`: Region code mapped from a known production `server_id`. + Current values are `AS`, `NA_SA`, `EU`, and `SE`. It is omitted for an + unrecognized server ID rather than guessed. - `records`: Pull/reward records. ## Fields to identify pulls diff --git a/pyproject.toml b/pyproject.toml index edfd7cd..f87550c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "nte-history-exporter" -version = "0.1.11" +version = "0.1.12" description = "Cross-platform Neverness to Everness pull-history exporter." requires-python = ">=3.10" dependencies = [] diff --git a/src/nte_history_exporter/__init__.py b/src/nte_history_exporter/__init__.py index 0c5c300..74acd0e 100644 --- a/src/nte_history_exporter/__init__.py +++ b/src/nte_history_exporter/__init__.py @@ -1 +1 @@ -__version__ = "0.1.11" +__version__ = "0.1.12" diff --git a/src/nte_history_exporter/adapters/mitmproxy_flows.py b/src/nte_history_exporter/adapters/mitmproxy_flows.py index 3cb8d48..b7500ac 100644 --- a/src/nte_history_exporter/adapters/mitmproxy_flows.py +++ b/src/nte_history_exporter/adapters/mitmproxy_flows.py @@ -12,6 +12,7 @@ from nte_history_exporter.decoder.mystery_box import ( select_continuous_mystery_box_run, ) from nte_history_exporter.decoder.user_uid import extract_user_uid_candidates +from nte_history_exporter.decoder.server_region import extract_server_id from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket @@ -84,10 +85,15 @@ 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() + server_id_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])) + server_id = extract_server_id(msg[1]) + if server_id: + server_id_candidates.update([server_id]) user_uid = user_uid_candidates.most_common(1)[0][0] if user_uid_candidates else None + server_id = server_id_candidates.most_common(1)[0][0] if server_id_candidates else None resolved_flow_index, flow = find_udp_flow(flows, flow_index) messages = flow[b"messages"] @@ -136,5 +142,6 @@ def decode_mitmproxy_flows(path: str | Path, flow_index: int | None = None) -> d "mystery_box_rows": mystery_box_rows, "mystery_box_warnings": mystery_box_warnings, "user_uid": session.user_uid or user_uid, + "server_id": session.server_id or server_id, "capture_diagnostics": session.diagnostic_report(), } diff --git a/src/nte_history_exporter/cli.py b/src/nte_history_exporter/cli.py index d5d4bf0..a02ed86 100644 --- a/src/nte_history_exporter/cli.py +++ b/src/nte_history_exporter/cli.py @@ -57,6 +57,7 @@ def main(argv: list[str] | None = None) -> int: write_debug_csv=args.debug, user_uid=args.user_uid, ) + console.wait_for_close() return 0 except (LibpcapUnavailable, PermissionError) as exc: console.print_problem(str(exc)) @@ -88,6 +89,9 @@ def main(argv: list[str] | None = None) -> int: resolved_user_uid = args.user_uid or decoded.get("user_uid") if not resolved_user_uid: resolved_user_uid = console.prompt_user_uid() + resolved_server_id = decoded.get("server_id") + if not resolved_server_id: + resolved_server_id = console.prompt_server_id() out_path, json_path = export_paths(kind, resolved_user_uid) diagnostics_path = None @@ -101,6 +105,7 @@ def main(argv: list[str] | None = None) -> int: source="packet_capture", capture_source="mitmproxy_flows", user_uid=resolved_user_uid, + server_id=resolved_server_id, flow_index=decoded["flow_index"], candidate_request_response_pairs=pair_count, pages_seen=[p[0] for p in best_run], diff --git a/src/nte_history_exporter/console.py b/src/nte_history_exporter/console.py index ecc65ba..10bb086 100644 --- a/src/nte_history_exporter/console.py +++ b/src/nte_history_exporter/console.py @@ -4,6 +4,8 @@ import os import sys from nte_history_exporter.constants import EXPORTER_VERSION, GAME_NAME +from nte_history_exporter.decoder.server_region import SERVER_REGIONS +from nte_history_exporter.live_capture.stop_key import wait_for_keypress WIDTH = 58 REPOSITORY_URL = "https://github.com/Golumpa/nte-exporter" @@ -179,3 +181,30 @@ def prompt_user_uid() -> str | None: print_note("Leaving this blank may prevent import on some trackers.") value = input(" User UID: ").strip() return value or None + + +def prompt_server_id() -> str | None: + print() + print_problem("Account server was not detected in this capture.") + print_note("Choose the server used by this account:") + choices = tuple( + (str(index), server_id, details["name"]) + for index, (server_id, details) in enumerate(SERVER_REGIONS.items(), start=1) + ) + for key, _server_id, name in choices: + print(f" {key}. {name}") + print_note("Leave this blank to omit server information.") + by_key = {key: server_id for key, server_id, _name in choices} + while True: + value = input(" Server [1-4]: ").strip() + if not value: + return None + if value in by_key: + return by_key[value] + print_problem("Enter 1, 2, 3, or 4; or leave it blank.") + + +def wait_for_close() -> None: + print() + print_success("Press any key to close the exporter.") + wait_for_keypress() diff --git a/src/nte_history_exporter/constants.py b/src/nte_history_exporter/constants.py index 300096e..0491f89 100644 --- a/src/nte_history_exporter/constants.py +++ b/src/nte_history_exporter/constants.py @@ -13,7 +13,7 @@ ARC_BANNER_NAME = "Arc Miracle Box" MYSTERY_BOX_BANNER_ID = "Gashapon_MysteryBox" MYSTERY_BOX_BANNER_NAME = "Mystery Box" EXPORTER_NAME = "nte-history-exporter" -EXPORTER_VERSION = "0.1.11" +EXPORTER_VERSION = "0.1.12" HISTORY_REQUEST_BANNER = 4220 HISTORY_REQUEST_LENGTH = 45 diff --git a/src/nte_history_exporter/decoder/server_region.py b/src/nte_history_exporter/decoder/server_region.py new file mode 100644 index 0000000..8106a58 --- /dev/null +++ b/src/nte_history_exporter/decoder/server_region.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import ipaddress +import struct + + +# Production entries from clientRes/OB-NTE-ServerList/serverlist_hhOS.json. +# Test and certification servers are intentionally excluded. +SERVER_REGIONS = { + "23001": {"account_region": "AS", "name": "Asia"}, + "23002": {"account_region": "NA_SA", "name": "America"}, + "23003": {"account_region": "EU", "name": "Europe"}, + "23004": {"account_region": "SE", "name": "SEA"}, +} + + +def account_region_for_server(server_id: str | None) -> str | None: + if not server_id: + return None + server = SERVER_REGIONS.get(str(server_id).strip()) + return server["account_region"] if server else None + + +def extract_server_id(payload: bytes) -> str | None: + """Extract DistrictId from the initial server-selection TCP response. + + This response is a length-prefixed FlatBuffers message. DistrictId is at + byte 96 in the current schema. Validate the frame and its following server + address before accepting the value so an arbitrary integer at that offset + is not mistaken for a server ID. + """ + if len(payload) < 168: + return None + + frame_size = struct.unpack_from(" len(payload) or frame_size < 168: + return None + if struct.unpack_from(" frame_size: + return None + try: + address = payload[136:address_end].decode("ascii") + ipaddress.ip_address(address) + except (UnicodeDecodeError, ValueError): + return None + + return str(server_id) diff --git a/src/nte_history_exporter/export/json_export.py b/src/nte_history_exporter/export/json_export.py index dd42d1f..774eef9 100644 --- a/src/nte_history_exporter/export/json_export.py +++ b/src/nte_history_exporter/export/json_export.py @@ -11,6 +11,7 @@ from nte_history_exporter.constants import ( POOL_META, MYSTERY_BOX_BANNER_ID, ) +from nte_history_exporter.decoder.server_region import account_region_for_server def build_export_json( @@ -20,6 +21,7 @@ def build_export_json( source: str = "packet_capture", capture_source: str | None = None, user_uid: str | None = None, + server_id: str | None = None, flow_index: int | None = None, candidate_request_response_pairs: int | None = None, pages_seen: list[int] | None = None, @@ -44,6 +46,7 @@ def build_export_json( scan["pages_seen"] = pages_seen normalized_user_uid = user_uid.strip() if user_uid else "" + normalized_server_id = str(server_id).strip() if server_id else "" export: dict[str, Any] = { "format": "nte-history-export", "format_version": 1, @@ -66,6 +69,11 @@ def build_export_json( ) if normalized_user_uid: export["user_uid"] = normalized_user_uid + if normalized_server_id: + export["server_id"] = normalized_server_id + account_region = account_region_for_server(normalized_server_id) + if account_region: + export["account_region"] = account_region export["records"] = [_record_for_export(r) for r in exported] return export diff --git a/src/nte_history_exporter/live_capture/runner.py b/src/nte_history_exporter/live_capture/runner.py index 4c38bc9..06e0914 100644 --- a/src/nte_history_exporter/live_capture/runner.py +++ b/src/nte_history_exporter/live_capture/runner.py @@ -144,6 +144,9 @@ def run_live_capture( 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() + resolved_server_id = session.server_id + if session.kinds_seen() and not resolved_server_id: + resolved_server_id = console.prompt_server_id() capture_source = CAPTURE_SOURCE_LABELS.get(capture.name, capture.name) for kind in session.kinds_seen(): pairs = session.pairs_for_kind(kind) @@ -162,6 +165,7 @@ def run_live_capture( source="live_capture", capture_source=capture_source, user_uid=resolved_user_uid, + server_id=resolved_server_id, pages_seen=pages_seen, ) payload = json.dumps(export, ensure_ascii=False, indent=2) diff --git a/src/nte_history_exporter/live_capture/session.py b/src/nte_history_exporter/live_capture/session.py index 5acaae3..0187654 100644 --- a/src/nte_history_exporter/live_capture/session.py +++ b/src/nte_history_exporter/live_capture/session.py @@ -30,6 +30,7 @@ from nte_history_exporter.decoder.mystery_box import ( from nte_history_exporter.decoder.structured_protocol import FORK_MARKER, MONOPOLY_MARKER from nte_history_exporter.constants import MYSTERY_BOX_MARKER from nte_history_exporter.decoder.user_uid import extract_user_uid_candidates +from nte_history_exporter.decoder.server_region import extract_server_id from nte_history_exporter.live_capture.diagnostics import CaptureDiagnostics @@ -72,6 +73,7 @@ class LiveHistorySession: self.unanswered_pages: dict[str, dict[int, str]] = {} self.user_uid: str | None = None self.user_uid_candidates: Counter[str] = Counter() + self.server_id: str | None = None self.diagnostics = CaptureDiagnostics() def _mark_unanswered(self, request: PendingRequest) -> None: @@ -122,6 +124,11 @@ class LiveHistorySession: if candidates: self.user_uid_candidates.update(candidates) self.user_uid = self.user_uid_candidates.most_common(1)[0][0] + if packet.protocol == "tcp" and packet.dst_ip == self.local_ip and not self.server_id: + self.server_id = extract_server_id(packet.payload) + if self.server_id: + self.diagnostics.counters["server_ids_detected"] += 1 + self.diagnostics.add_event("SERVER_ID_DETECTED", self.packet_count) if packet.protocol != "udp": return False diff --git a/src/nte_history_exporter/live_capture/stop_key.py b/src/nte_history_exporter/live_capture/stop_key.py index c97f2cd..c6e1daa 100644 --- a/src/nte_history_exporter/live_capture/stop_key.py +++ b/src/nte_history_exporter/live_capture/stop_key.py @@ -5,6 +5,41 @@ import select import sys +def _read_windows_key() -> None: + import msvcrt + + key = msvcrt.getch() + # Function and arrow keys are emitted as a two-byte sequence. Consume the + # second byte so it cannot immediately satisfy the next key prompt. + if key in {b"\x00", b"\xe0"}: + msvcrt.getch() + + +def wait_for_keypress() -> None: + """Wait for one key on an interactive terminal, or Enter on redirected input.""" + if os.name == "nt" and sys.stdin.isatty(): + _read_windows_key() + return + + if not sys.stdin.isatty(): + try: + input() + except EOFError: + pass + return + + import termios + import tty + + fd = sys.stdin.fileno() + original_terminal = termios.tcgetattr(fd) + try: + tty.setcbreak(fd) + os.read(fd, 1) + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, original_terminal) + + class StopKeyMonitor: def __init__(self) -> None: self._windows = os.name == "nt" @@ -29,7 +64,7 @@ class StopKeyMonitor: if not msvcrt.kbhit(): return False - msvcrt.getch() + _read_windows_key() return True if self._fd is None: diff --git a/tests/support.py b/tests/support.py index 58a9a55..122df15 100644 --- a/tests/support.py +++ b/tests/support.py @@ -21,6 +21,7 @@ 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.server_region import extract_server_id from nte_history_exporter.export.csv_export import write_csv from nte_history_exporter.decoder.arc import ( arc_request_page, @@ -46,6 +47,7 @@ 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 import console from nte_history_exporter.pool_mappings import load_pool_mappings, pool_meta_from_mapping from nte_history_exporter.update_check import UpdateInfo, check_for_update, is_newer_version diff --git a/tests/test_export_contract.py b/tests/test_export_contract.py index 24cc919..24441ed 100644 --- a/tests/test_export_contract.py +++ b/tests/test_export_contract.py @@ -10,6 +10,8 @@ class ExportContractTests(unittest.TestCase): self.assertEqual(export["format"], "nte-history-export") self.assertIn("exporter", export) self.assertNotIn("user_uid", export) + self.assertNotIn("server_id", export) + self.assertNotIn("account_region", export) self.assertNotIn("record_hex", export["records"][0]) self.assertNotIn("request_msg", export["records"][0]) self.assertNotIn("response_msg", export["records"][0]) @@ -27,6 +29,75 @@ class ExportContractTests(unittest.TestCase): self.assertEqual(export["capture_source"], "npcap") self.assertEqual(export["user_uid"], "123456789") + def test_export_includes_server_id_and_mapped_account_region(self): + annotated = annotate_groups(fixture_session().build_rows("permanent")) + export = build_export_json(annotated, [], server_id="23003") + + self.assertEqual(export["server_id"], "23003") + self.assertEqual(export["account_region"], "EU") + + def test_export_preserves_unknown_server_without_guessing_region(self): + annotated = annotate_groups(fixture_session().build_rows("permanent")) + export = build_export_json(annotated, [], server_id="23999") + + self.assertEqual(export["server_id"], "23999") + self.assertNotIn("account_region", export) + + def test_extracts_server_id_from_valid_initial_tcp_response(self): + payload = bytearray(204) + payload[0:4] = (200).to_bytes(4, "little") + payload[4:8] = (20).to_bytes(4, "little") + payload[96:100] = (23003).to_bytes(4, "little") + address = b"198.51.100.20" + payload[132:136] = len(address).to_bytes(4, "little") + payload[136 : 136 + len(address)] = address + + self.assertEqual(extract_server_id(bytes(payload)), "23003") + + def test_rejects_server_id_at_offset_without_valid_message_structure(self): + payload = bytearray(204) + payload[96:100] = (23003).to_bytes(4, "little") + + self.assertIsNone(extract_server_id(bytes(payload))) + + def test_live_session_detects_server_only_on_inbound_tcp(self): + payload = bytearray(204) + payload[0:4] = (200).to_bytes(4, "little") + payload[4:8] = (20).to_bytes(4, "little") + payload[96:100] = (23004).to_bytes(4, "little") + address = b"198.51.100.20" + payload[132:136] = len(address).to_bytes(4, "little") + payload[136 : 136 + len(address)] = address + session = LiveHistorySession("192.0.2.10") + + session.process_packet( + UdpPacket( + 1.0, + "198.51.100.20", + "192.0.2.10", + 30000, + 40000, + bytes(payload), + protocol="tcp", + ) + ) + + self.assertEqual(session.server_id, "23004") + + @patch("builtins.input", side_effect=["9", "3"]) + def test_server_prompt_retries_then_returns_selected_server_id(self, _input): + self.assertEqual(console.prompt_server_id(), "23003") + + @patch("builtins.input", return_value="") + def test_server_prompt_can_be_skipped(self, _input): + self.assertIsNone(console.prompt_server_id()) + + @patch("nte_history_exporter.console.wait_for_keypress") + def test_wait_for_close_waits_for_a_second_keypress(self, wait_for_keypress): + console.wait_for_close() + + wait_for_keypress.assert_called_once_with() + def test_debug_csv_includes_exporter_version(self): with TemporaryDirectory() as tmp: path = Path(tmp) / "debug.csv"