cc88955baf
Implemented server detection at the request of NTEWIZ. If not detected, it will ask after collecting pull data.
179 lines
6.9 KiB
Python
179 lines
6.9 KiB
Python
from tests.support import * # noqa: F401,F403
|
|
from nte_history_exporter import __version__
|
|
|
|
|
|
class ExportContractTests(unittest.TestCase):
|
|
def test_sanitized_export_omits_raw_packet_fields(self):
|
|
annotated = annotate_groups(fixture_session().build_rows("permanent"))
|
|
export = build_export_json(annotated, [])
|
|
|
|
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])
|
|
|
|
def test_export_includes_user_uid_when_provided(self):
|
|
annotated = annotate_groups(fixture_session().build_rows("permanent"))
|
|
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_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"
|
|
write_csv(path, [{"uid": "abc123"}])
|
|
|
|
with path.open(newline="", encoding="utf-8") as f:
|
|
rows = list(csv.DictReader(f))
|
|
|
|
self.assertEqual(rows[0]["exporter_version"], __version__)
|
|
self.assertEqual(rows[0]["uid"], "abc123")
|
|
|
|
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)
|