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

feat: capture and export gear, cartridges and loadouts

Browse source 
Adds a third export alongside pull history and achievements: the cartridges,
modules, character levels and equipped console boards the game sends at login.

Shaped as added files plus a handful of one-line calls, so this repo can keep
merging upstream:

- decoder/gear.py: self-contained bit-level decoder over a list of datagrams.
- export/gear_export.py: the nte-gear-export envelope.
- live_capture/gear_collector.py: UDP flow accumulation, the mid-capture
  decode, the fallback scan, the console line and the export file.

Gear is reported during the capture rather than only at the stop. The burst
arrives in a rush and then stops, so "this flow went quiet" is the only
mid-stream marker available; a flow is retried only once it has grown, since
decoding per packet would be quadratic in flow size.

Flows are tried largest first and the first one that decodes wins - picking
the single biggest was wrong, history traffic can outweigh a short login
burst.

Verified against a live capture: 817 items (310 cartridges, 507 modules),
20 characters, 13 loadouts, every bucket matching the in-game counts exactly.
26 synthetic tests, no real capture used as a fixture; 149 total.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
goober 2026-08-21 11:38:27 +03:00
parent 93c6626640
commit 3c8834e98f
10 changed files with 1568 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

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

2
src/nte_history_exporter/__init__.py
View file

@ -1 +1 @@
__version__ = "0.3.0"
__version__ = "0.4.0"

4
src/nte_history_exporter/console.py
View file

@ -97,6 +97,10 @@ def print_live_instructions(local_ip: str, backend: str = "windows_raw", detail:
print(" Your tracked achievements are captured automatically when")
print(" you log in after following the step above.")
print()
print(style(" Gear", BOLD))
print(" Your cartridges, modules and equipped loadouts are captured")
print(" automatically at login, in the same step as achievements.")
print()
print(style(" Ready - use the game, then press any key here when finished.", BOLD, GREEN))
print(rule())

2
src/nte_history_exporter/constants.py
View file

@ -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.3.0"
EXPORTER_VERSION = "0.4.0"
HISTORY_REQUEST_BANNER = 4220
HISTORY_REQUEST_LENGTH = 45

499
src/nte_history_exporter/decoder/gear.py Normal file
View file

@ -0,0 +1,499 @@
"""Decode cartridges, modules and character loadouts from the gameplay UDP flow.
The game speaks Unreal's bit-packed net protocol. Each datagram is a bit stream:
a 15-byte packet and bunch header, the payload, then two 1 bits - the packet
handler's terminator and the bit writer's - zero-padded to the next byte
boundary. Records routinely straddle datagram boundaries, so the payload bit
ranges have to be spliced; concatenating whole datagrams splices the framing into
the middle of a record and corrupts it.
Inside the spliced stream, an item record reads:
<u32 length> length == len(id) + 1 (self-validating frame check)
<ascii id>\0
<5 bytes> <8-byte instance id> ... <u32 level at +38 past the id>
... <12-byte owner reference at +312 bits, all zero when unequipped> ...
then N stat entries, each:
<u32 length> <ascii stat name>\0 <4 zero bytes> <f32 value>
Consecutive entries drift one bit apart, so entries are chained forward from the
previous one rather than found by global search - that way a neighbouring
record's entries can never be stolen.
A character block is marked by its `GA_<Name>_Melee|Skill|UltraSkill|QTE` ability
strings, with the character's level and breakthrough count at fixed bit offsets
before it, and its equipped modules listed afterwards one entry per occupied
grid cell.
Every failure path returns empty rather than raising.
"""
from __future__ import annotations
import re
import struct
from dataclasses import dataclass, field
# --- framing ---------------------------------------------------------------
HEADER_BITS = 120
# Datagrams with no room for a payload are pure acks and carry nothing.
MIN_PAYLOAD_BITS = 32
# --- record layout ---------------------------------------------------------
LEVEL_OFFSET_BYTES = 38
MAX_ITEM_LEVEL = 20
OWNER_OFFSET_BITS = 312
OWNER_BYTES = 12
# An item packs its entries tightly; anything further away belongs to something
# else, so a wider scan would wander into unrelated data.
CHAIN_SCAN_BITS = 1400
# Observed entry spacing is (len+12) bytes plus a 1-bit separator, tightest at
# ~145 bits. Advancing by a floor below that and rescanning avoids overshooting
# entries that are shorter than their computed width - main stats run two bytes
# shorter than substats.
MIN_ENTRY_SPACING_BITS = 100
MODULE_ENTRIES = 6
CARTRIDGE_ENTRIES = 5
# --- character layout ------------------------------------------------------
# Bit offsets relative to the start of the character's first ability frame.
CHARACTER_MARKER_OFFSET_BITS = -466
CHARACTER_LEVEL_OFFSET_BITS = -434
CHARACTER_BREAKTHROUGH_OFFSET_BITS = -402
CHARACTER_MARKER = 67806
MAX_CHARACTER_LEVEL = 90
MAX_BREAKTHROUGHS = 10
# A loadout cell entry stores doubled grid coordinates after the item id.
CELL_ROW_OFFSET_BITS = 96
CELL_COL_OFFSET_BITS = 128
# --- vocabulary ------------------------------------------------------------
# Ids are whitelisted: the length prefix validates a frame but cannot catch a
# single flipped bit inside the name - `Ingantation` and `GetEfviciency` both
# passed the length check before the whitelist went in.
ITEM_ID = re.compile(
rb"(cell[234]_style[1-6]_\d_(?:Orange|Purple|Blue)"
# Psychically must precede Psyche: both ids exist and name different sets.
rb"|(?:Attack|Chaos|Cosmos|GetEfficiency|Heal|Incantation|Lakshana|Mag|Nature"
rb"|Psychically|Psyche|Shield)_(?:orange|purple|blue))\x00"
)
ABILITY_ID = re.compile(rb"GA_([A-Za-z0-9]+)_(?:Melee|Skill|UltraSkill|QTE)\x00")
STAT_NAMES = frozenset({
"AtkAdd", "AtkUp", "DefAdd", "DefUp", "HPMaxAdd", "HPMaxUp", "CritBase",
"CritDamageBase", "MagBase", "UnbalIntensityBase", "HealUp",
})
# Substat values are fixed per stat and item size, with no roll variance, so any
# value off this table means the parse went wrong. Main stats read 0.0.
CANONICAL_VALUES = frozenset({
0.0, 200.0, 300.0, 400.0, 1000.0, 16.0, 24.0, 32.0, 80.0, 48.0, 64.0, 600.0, 800.0,
12.0, 18.0, 60.0, 36.0,
0.025, 0.0375, 0.05, 0.125, 0.035, 0.0525, 0.07, 0.075, 0.175, 0.02, 0.03, 0.04,
0.1, 0.06, 0.08, 0.2, 0.105, 0.14,
})
# A main stat's displayed value is never transmitted: it is base * (1 + level/5),
# so +0 shows the base and +20 shows five times it.
CARTRIDGE_MAIN_BASE = {
"HPMaxUp": 0.075, "AtkUp": 0.075, "DefUp": 0.105, "CritBase": 0.06,
"CritDamageBase": 0.12, "MagBase": 36.0, "UnbalIntensityBase": 36.0,
"DamageUpGeneralBase": 0.075,
# Healing Bonus reads 6.90% at +0 in game - the one main that is not round.
"HealUp": 0.069,
}
# Module mains scale per cell: 56 HP and 4.2 ATK each at +0. Measured in game as
# Type III 168/12.6 -> 840/63 and Type IV 224/16.8 -> 1120/84. The client floors
# flat values, which is why ATK displays as 12 and 16 at +0.
MODULE_HP_PER_CELL = 56.0
MODULE_ATK_PER_CELL = 4.2
# Confirmed against everness's cartridge boxes, which name all twelve sets. The
# three the capture has never carried are Psyche, Shield and Heal - NOT Blood,
# Night and Kingdom, which were a guess and would have decoded as nothing.
SET_NAMES = {
"Attack": "Shadow Creed", "Chaos": "Diabolos", "Cosmos": "Lost Radiance",
"GetEfficiency": "Speedy Hedgehog", "Incantation": "Crimson: Twin Butterflies",
"Lakshana": "Street Boxer", "Nature": "Fireflies and the Forest",
"Psychically": "Quiet Manor", "Mag": "Tiny Big Adventure",
"Psyche": "Devil's Blood: Curse", "Heal": "Thea's Night Tavern",
"Shield": "Kingdom's Guard",
}
MODULE_TYPE_NAMES = {2: "II", 3: "III", 4: "IV"}
@dataclass(frozen=True)
class GearItem:
instance: str
kind: str
item_id: str
level: int
rarity: str
main_stats: tuple[tuple[str, float | None], ...]
substats: tuple[tuple[str, float], ...]
owner_group: str | None
shape: str | None = None
module_type: str | None = None
set_name: str | None = None
@dataclass(frozen=True)
class GearCharacter:
key: str
level: int | None
breakthroughs: int | None
board: tuple[tuple[int, int], ...] = ()
@dataclass(frozen=True)
class GearSnapshot:
items: tuple[GearItem, ...] = ()
characters: tuple[GearCharacter, ...] = ()
warnings: tuple[str, ...] = field(default=())
@property
def cartridges(self) -> int:
return sum(item.kind == "cartridge" for item in self.items)
@property
def modules(self) -> int:
return sum(item.kind == "module" for item in self.items)
@property
def characters_with_loadouts(self) -> int:
return sum(bool(character.board) for character in self.characters)
def is_stat_name(name: str) -> bool:
return name in STAT_NAMES or (name.startswith("DamageUp") and name.endswith("Base"))
def bitshift(buf: bytes, shift: int) -> bytes:
if shift == 0:
return buf
return bytes(
((buf[i] >> shift) | (buf[i + 1] << (8 - shift))) & 0xFF
for i in range(len(buf) - 1)
)
class _BitAccumulator:
"""Append arbitrary bit ranges; the result is a byte buffer, LSB-first."""
def __init__(self) -> None:
self.buf = bytearray()
self.nbits = 0
def append(self, data: bytes, start: int, nbits: int) -> None:
if nbits <= 0:
return
value = (int.from_bytes(data, "little") >> start) & ((1 << nbits) - 1)
offset = self.nbits % 8
if offset:
value <<= offset
raw = value.to_bytes((offset + nbits + 7) // 8, "little")
self.buf[-1] |= raw[0]
self.buf.extend(raw[1:])
else:
self.buf.extend(value.to_bytes((nbits + 7) // 8, "little"))
self.nbits += nbits
del self.buf[(self.nbits + 7) // 8 :]
def payload_end_bit(datagram: bytes) -> int:
"""First bit past the payload.
Unreal ends a packet with two 1 bits then zero-pads to a byte boundary, so
the payload stops one bit below the highest set bit of the last non-zero
byte.
"""
i = len(datagram) - 1
while i >= 0 and datagram[i] == 0:
i -= 1
if i < 0:
return 0
return i * 8 + datagram[i].bit_length() - 2
def splice_datagrams(datagrams) -> bytes:
"""Concatenate every datagram's payload bits with the framing removed."""
accumulator = _BitAccumulator()
for datagram in datagrams:
end = payload_end_bit(datagram)
if end - HEADER_BITS >= MIN_PAYLOAD_BITS:
accumulator.append(datagram, HEADER_BITS, end - HEADER_BITS)
return bytes(accumulator.buf)
class _BitView:
"""Random access to a buffer at arbitrary bit offsets."""
def __init__(self, buf: bytes) -> None:
self.views = [bitshift(buf, shift) for shift in range(8)]
def at(self, bitpos: int) -> tuple[bytes, int]:
return self.views[bitpos % 8], bitpos // 8
def u32(self, bitpos: int) -> int | None:
view, offset = self.at(bitpos)
if offset + 4 > len(view):
return None
return struct.unpack_from("<I", view, offset)[0]
def f32(self, bitpos: int) -> float | None:
view, offset = self.at(bitpos)
if offset + 4 > len(view):
return None
return struct.unpack_from("<f", view, offset)[0]
def raw(self, bitpos: int, length: int) -> bytes:
view, offset = self.at(bitpos)
return view[offset : offset + length]
def text(self, bitpos: int, length: int) -> str | None:
view, offset = self.at(bitpos)
if offset + length > len(view):
return None
raw = view[offset : offset + length]
if not raw.endswith(b"\0"):
return None
try:
text = raw[:-1].decode("ascii")
except UnicodeDecodeError:
return None
return text if text.isprintable() else None
def _validated_frames(view: _BitView, pattern):
"""Every self-validating <u32 len><name\\0> frame, in bit order."""
found = []
for shift in range(8):
haystack = view.views[shift]
for match in pattern.finditer(haystack):
body = match.group()
if match.start() < 4:
continue
if struct.unpack_from("<I", haystack, match.start() - 4)[0] != len(body):
continue
found.append((
(match.start() - 4) * 8 + shift, # frame start
match,
shift,
match.end() * 8 + shift, # first bit past the name
))
found.sort()
return found
def read_entry(view: _BitView, bitpos: int) -> tuple[str, float] | None:
"""Read one length-prefixed <name, f32> stat entry."""
declared = view.u32(bitpos)
if declared is None or not 4 <= declared <= 64:
return None
name = view.text(bitpos + 32, declared)
if name is None:
return None
# Layout: <u32 len><name\0><4 zero bytes><f32 value>
value = view.f32(bitpos + 32 + declared * 8 + 32)
if value is None:
return None
return name, round(value, 6)
def _chain_entries(view: _BitView, start_bit: int, needed: int):
"""Walk `needed` stat entries forward from just past the item id."""
cursor, entries = start_bit, []
while len(entries) < needed:
found = None
for delta in range(CHAIN_SCAN_BITS):
probe = read_entry(view, cursor + delta)
if probe and is_stat_name(probe[0]):
found = (probe, cursor + delta)
break
if not found:
return entries
entries.append(found[0])
cursor = found[1] + MIN_ENTRY_SPACING_BITS
return entries
def _main_value(kind: str, stat: str, level: int, cells: int | None) -> float | None:
if kind == "module":
base = {
"HPMaxAdd": MODULE_HP_PER_CELL * (cells or 0),
"AtkAdd": MODULE_ATK_PER_CELL * (cells or 0),
}.get(stat)
else:
base = CARTRIDGE_MAIN_BASE.get(stat)
# Every elemental damage bonus shares the general bonus's base.
if base is None and stat.startswith("DamageUp") and stat.endswith("Base"):
base = CARTRIDGE_MAIN_BASE["DamageUpGeneralBase"]
if base is None:
return None
value = base * (1 + level / 5)
# Flat module stats display floored; percentages keep their precision.
return float(int(value)) if kind == "module" else round(value, 4)
def _item_from_frame(view: _BitView, frame) -> GearItem | None:
bitpos, match, shift, id_end_bit = frame
item_id = match.group()[:-1].decode()
is_module = item_id.startswith("cell")
needed = MODULE_ENTRIES if is_module else CARTRIDGE_ENTRIES
main_count = 2 if is_module else 1
entries = _chain_entries(view, id_end_bit, needed)
if len(entries) != needed:
return None
# Substats never vary, so an off-table value means the chain went astray.
if not all(value in CANONICAL_VALUES for _, value in entries):
return None
# In every record confirmed against in-game values the main stats come
# first. A main appearing later means the chain wrapped into a neighbour.
if not all(value == 0.0 for _, value in entries[:main_count]):
return None
if sum(value == 0.0 for _, value in entries) != main_count:
return None
id_end_byte = match.end()
haystack = view.views[shift]
instance = haystack[id_end_byte + 5 : id_end_byte + 13]
if len(instance) != 8:
return None
level_offset = id_end_byte + LEVEL_OFFSET_BYTES
if level_offset + 4 > len(haystack):
return None
level = struct.unpack_from("<I", haystack, level_offset)[0]
if level > MAX_ITEM_LEVEL:
level = 0
owner_bit = id_end_bit + OWNER_OFFSET_BITS
# Gate on the leading word: trailing bytes of the field carry unrelated data
# on some records, so testing the whole span over-reports.
owner = view.raw(owner_bit, OWNER_BYTES) if view.u32(owner_bit) else b""
cells = int(item_id[4]) if is_module else None
mains = tuple(
(name, _main_value("module" if is_module else "cartridge", name, level, cells))
for name, value in entries if value == 0.0
)
substats = tuple((name, value) for name, value in entries if value != 0.0)
rarity = item_id.rsplit("_", 1)[1].lower()
if is_module:
shape = "_".join(item_id.split("_")[:2])
return GearItem(
instance=instance.hex(), kind="module", item_id=item_id, level=level,
rarity=rarity, main_stats=mains, substats=substats,
owner_group=owner.hex() or None,
shape=shape, module_type=MODULE_TYPE_NAMES.get(cells or 0),
)
return GearItem(
instance=instance.hex(), kind="cartridge", item_id=item_id, level=level,
rarity=rarity, main_stats=mains, substats=substats,
owner_group=owner.hex() or None,
set_name=SET_NAMES.get(item_id.rsplit("_", 1)[0]),
)
def _character_blocks(view: _BitView) -> list[tuple[int, str]]:
"""The first ability frame of each character, in stream order."""
blocks: list[tuple[int, str]] = []
for bitpos, match, _shift, _end in _validated_frames(view, ABILITY_ID):
name = match.group(1).decode()
if not blocks or blocks[-1][1] != name:
blocks.append((bitpos, name))
return blocks
def _character_progress(view: _BitView, block_bit: int) -> tuple[int | None, int | None]:
"""Level and breakthrough count, validated against the block's marker."""
if view.u32(block_bit + CHARACTER_MARKER_OFFSET_BITS) != CHARACTER_MARKER:
return None, None
level = view.u32(block_bit + CHARACTER_LEVEL_OFFSET_BITS)
breakthroughs = view.u32(block_bit + CHARACTER_BREAKTHROUGH_OFFSET_BITS)
if level is None or not 1 <= level <= MAX_CHARACTER_LEVEL:
level = None
if breakthroughs is None or breakthroughs > MAX_BREAKTHROUGHS:
breakthroughs = None
return level, breakthroughs
def extract_gear(datagrams) -> GearSnapshot:
"""Decode every item and character loadout in a captured UDP flow."""
try:
buf = splice_datagrams(datagrams)
except (ValueError, MemoryError):
return GearSnapshot()
if len(buf) < 64:
return GearSnapshot()
view = _BitView(buf)
frames = _validated_frames(view, ITEM_ID)
blocks = _character_blocks(view)
items: dict[str, GearItem] = {}
# An id frame that chains no stat entries is a loadout reference, not a
# record: the same id quoted inside a character's equipped list.
loadouts: dict[str, list[tuple[int, int, int]]] = {}
for frame in frames:
item = _item_from_frame(view, frame)
if item is not None:
items.setdefault(item.instance, item)
continue
bitpos, _match, _shift, id_end_bit = frame
row = view.u32(id_end_bit + CELL_ROW_OFFSET_BITS)
column = view.u32(id_end_bit + CELL_COL_OFFSET_BITS)
if not row or not column:
continue
owner_index = -1
for index, (block_bit, _name) in enumerate(blocks):
if block_bit > bitpos:
break
owner_index = index
if owner_index >= 0:
# Coordinates are doubled; a cell is one square, not four.
loadouts.setdefault(blocks[owner_index][1], []).append(
(bitpos, row // 2, column // 2)
)
# A character's block is transmitted more than once in a login - the active
# team appears early and again in the full roster - so merge by name and
# keep the first reading that validated.
merged: dict[str, GearCharacter] = {}
for block_bit, name in blocks:
level, breakthroughs = _character_progress(view, block_bit)
board = tuple(sorted({(r, c) for _bit, r, c in loadouts.get(name, [])}))
previous = merged.get(name)
if previous is None:
merged[name] = GearCharacter(name, level, breakthroughs, board)
continue
merged[name] = GearCharacter(
key=name,
level=previous.level if previous.level is not None else level,
breakthroughs=(
previous.breakthroughs
if previous.breakthroughs is not None
else breakthroughs
),
board=previous.board or board,
)
return GearSnapshot(
items=tuple(items.values()),
characters=tuple(merged.values()),
)

99
src/nte_history_exporter/export/gear_export.py Normal file
View file

@ -0,0 +1,99 @@
"""Gear export envelope.
Kept out of ``json_export`` so the gear feature is a set of added files rather
than edits to upstream ones - this repo tracks Golumpa's exporter and has to be
able to merge it.
"""
from __future__ import annotations
from typing import Any
from nte_history_exporter import __version__
from nte_history_exporter.constants import EXPORTER_NAME, GAME_NAME
from nte_history_exporter.decoder.gear import GearCharacter, GearItem, GearSnapshot
from nte_history_exporter.decoder.server_region import account_region_for_server
def build_gear_export_json(
snapshot: GearSnapshot,
*,
source: str = "packet_capture",
capture_source: str | None = None,
user_uid: str | None = None,
server_id: str | None = None,
) -> dict[str, Any]:
scan = {
"cartridges": snapshot.cartridges,
"modules": snapshot.modules,
"characters": len(snapshot.characters),
"characters_with_loadouts": snapshot.characters_with_loadouts,
"warnings": list(snapshot.warnings),
}
export: dict[str, Any] = {
"format": "nte-gear-export",
"format_version": 1,
"game": GAME_NAME,
"source": source,
}
if capture_source:
export["capture_source"] = capture_source
export["exporter"] = {"name": EXPORTER_NAME, "version": __version__}
export["scan"] = scan
normalized_user_uid = user_uid.strip() if user_uid else ""
normalized_server_id = str(server_id).strip() if server_id else ""
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["characters"] = [
_gear_character_for_export(character) for character in snapshot.characters
]
export["items"] = [_gear_item_for_export(item) for item in snapshot.items]
return export
def _gear_character_for_export(character: GearCharacter) -> dict[str, Any]:
record: dict[str, Any] = {
"key": character.key,
"level": character.level,
"breakthroughs": character.breakthroughs,
}
# Only emitted when a loadout was actually seen; an empty board and an
# unequipped character are different states and must not collapse.
if character.board:
record["board"] = [[row, column] for row, column in character.board]
# The capture never ties an owner group to a character name, so the app
# resolves this itself rather than being handed a guess.
record["owner_group"] = None
return record
def _gear_item_for_export(item: GearItem) -> dict[str, Any]:
record: dict[str, Any] = {
"instance": item.instance,
"kind": item.kind,
"item_id": item.item_id,
}
if item.shape:
record["shape"] = item.shape
if item.module_type:
record["module_type"] = item.module_type
if item.set_name:
record["set"] = item.set_name
record["level"] = item.level
record["rarity"] = item.rarity
record["main_stats"] = [
{"stat": stat, "value": value} for stat, value in item.main_stats
]
record["substats"] = [
{"stat": stat, "value": value} for stat, value in item.substats
]
record["owner_group"] = item.owner_group
return record

191
src/nte_history_exporter/live_capture/gear_collector.py Normal file
View file

@ -0,0 +1,191 @@
"""Gear capture, held in one object so ``runner`` only has to call it.
This repo tracks Golumpa's exporter, so the gear feature is deliberately shaped
as added files plus a handful of one-line calls in ``runner.run_live_capture``
rather than as edits scattered through the capture loop. Everything gear needs -
flow accumulation, the mid-capture decode, the fallback scan, the console line
and the export file - lives here.
"""
from __future__ import annotations
import json
import time
from datetime import datetime
from pathlib import Path
from nte_history_exporter.console import BOLD, DIM, GREEN, style
from nte_history_exporter.decoder.gear import GearSnapshot, extract_gear
from nte_history_exporter.export.gear_export import build_gear_export_json
Flow = tuple[str, int, str, int]
# The login gear burst is roughly 500 KB; the cap only exists so a long
# session on a busy flow cannot grow without bound.
MAX_UDP_FLOW_BYTES = 8 * 1024 * 1024
# Decoding is linear in flow size, so only the few busiest are worth trying.
MAX_GEAR_FLOW_ATTEMPTS = 4
# A gear burst arrives in a rush and then stops, so "this flow has gone quiet"
# is the only mid-stream marker there is. Attempts are gated on the flow being
# large enough to hold a burst, having been idle, and having grown since it was
# last tried, so a quiet flow is never decoded twice for the same bytes.
GEAR_LIVE_MIN_FLOW_BYTES = 64 * 1024
GEAR_LIVE_IDLE_SECONDS = 1.0
GEAR_LIVE_MAX_ATTEMPTS = 8
def print_gear_captured(
cartridges: int,
modules: int,
characters: int,
characters_with_loadouts: int,
) -> None:
print(style(" + ", GREEN, BOLD) + "Gear captured")
print(
style(
f" Inventory {cartridges} cartridges, {modules} modules",
DIM,
)
)
print(
style(
f" Characters {characters} seen, "
f"{characters_with_loadouts} with loadouts",
DIM,
)
)
def gear_path(user_uid: str | None = None) -> Path:
export_dir = Path("exports")
export_dir.mkdir(parents=True, exist_ok=True)
uid_prefix = _safe_filename_part(user_uid) if user_uid else "unknown"
base = export_dir / f"{uid_prefix}_Gear_{datetime.now():%Y%m%d_%H%M%S}"
path = base.with_suffix(".json")
counter = 2
while path.exists():
path = export_dir / f"{base.name}_{counter}.json"
counter += 1
return path
def _safe_filename_part(value: str | None) -> str:
# Deliberately a copy of runner's helper rather than an import: importing
# runner from here would be circular, and the rule is four lines long.
if not value:
return "unknown"
cleaned = "".join(ch for ch in value.strip() if ch.isalnum() or ch in ("-", "_"))
return cleaned or "unknown"
class GearCollector:
"""Accumulates inbound UDP and decodes gear out of it.
Call ``poll`` once per capture iteration (above any ``packet is None``
guard - the pcap read timeout is what keeps it firing after the burst goes
quiet), ``observe`` for each packet, ``finalize`` after the loop, and
``write`` to emit the export.
"""
def __init__(self, local_ip: str | None) -> None:
self._local_ip = local_ip
self._flows: dict[Flow, list[bytes]] = {}
self._flow_bytes: dict[Flow, int] = {}
self._flow_last_seen: dict[Flow, float] = {}
self._attempted_bytes: dict[Flow, int] = {}
self._attempts = 0
self.snapshot: GearSnapshot | None = None
def observe(self, packet) -> None:
"""Keep an inbound UDP payload whole.
Datagram boundaries carry the framing, so payloads are spliced at
decode time and never concatenated here.
"""
if (
packet is None
or packet.protocol != "udp"
or not packet.payload
or packet.dst_ip != self._local_ip
):
return
flow: Flow = (packet.src_ip, packet.src_port, packet.dst_ip, packet.dst_port)
if self._flow_bytes.get(flow, 0) + len(packet.payload) > MAX_UDP_FLOW_BYTES:
return
self._flows.setdefault(flow, []).append(packet.payload)
self._flow_bytes[flow] = self._flow_bytes.get(flow, 0) + len(packet.payload)
self._flow_last_seen[flow] = time.monotonic()
def poll(self) -> None:
"""Decode a flow that has stopped growing, and report it immediately."""
if self.snapshot is not None or self._attempts >= GEAR_LIVE_MAX_ATTEMPTS:
return
now = time.monotonic()
candidates = [
flow
for flow, size in self._flow_bytes.items()
if size >= GEAR_LIVE_MIN_FLOW_BYTES
and self._attempted_bytes.get(flow) != size
and now - self._flow_last_seen.get(flow, now) >= GEAR_LIVE_IDLE_SECONDS
]
candidates.sort(key=self._flow_bytes.__getitem__, reverse=True)
for flow in candidates[:MAX_GEAR_FLOW_ATTEMPTS]:
self._attempted_bytes[flow] = self._flow_bytes[flow]
self._attempts += 1
snapshot = extract_gear(self._flows[flow])
if snapshot.items:
self._report(snapshot)
return
def finalize(self) -> None:
"""Fallback scan for a flow that never idled.
Normally ``poll`` has already decoded the burst. Rescanning per packet
instead of only here would be quadratic in flow size.
"""
if self.snapshot is not None:
return
# The gameplay flow dwarfs the others, but history traffic can outweigh
# a short login burst, so size only sets the order - the first flow that
# actually decodes wins.
busiest = sorted(self._flow_bytes, key=self._flow_bytes.__getitem__, reverse=True)
for flow in busiest[:MAX_GEAR_FLOW_ATTEMPTS]:
snapshot = extract_gear(self._flows[flow])
if snapshot.items:
self._report(snapshot)
return
def write(
self,
*,
capture_source: str | None = None,
user_uid: str | None = None,
server_id: str | None = None,
source: str = "live_capture",
) -> Path | None:
if self.snapshot is None:
return None
path = gear_path(user_uid)
export = build_gear_export_json(
self.snapshot,
source=source,
capture_source=capture_source,
user_uid=user_uid,
server_id=server_id,
)
path.write_text(
json.dumps(export, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
return path
def _report(self, snapshot: GearSnapshot) -> None:
self.snapshot = snapshot
print_gear_captured(
snapshot.cartridges,
snapshot.modules,
len(snapshot.characters),
snapshot.characters_with_loadouts,
)

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

@ -21,6 +21,7 @@ from nte_history_exporter.export.json_export import (
)
from nte_history_exporter.live_capture.backends import open_capture_backend
from nte_history_exporter.live_capture.diagnostics import new_diagnostics_path, write_capture_diagnostics
from nte_history_exporter.live_capture.gear_collector import GearCollector
from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket
from nte_history_exporter.live_capture.stop_key import StopKeyMonitor
from nte_history_exporter.live_capture.windows_raw import detect_local_ipv4
@ -88,14 +89,17 @@ def run_live_capture(
active_gap_notices: set[str] = set()
tcp_segments: dict[tuple[str, int, str, int], list[tuple[int, bytes]]] = {}
achievement_records = []
gear = GearCollector(local_ip)
try:
with StopKeyMonitor() as stop_key:
for packet in capture.packets():
if stop_key.pressed():
break
gear.poll() # above the guard: the pcap read timeout drives it
if packet is None:
continue
gear.observe(packet)
if (
not achievement_records
and packet.protocol == "tcp"
@ -176,6 +180,8 @@ def run_live_capture(
stats = capture.stats()
capture.close()
gear.finalize()
exports = []
achievement_path = None
diagnostics_path = None
@ -183,7 +189,9 @@ def run_live_capture(
diagnostics_path = new_diagnostics_path()
write_capture_diagnostics(diagnostics_path, session.diagnostic_report())
resolved_user_uid = user_uid or session.user_uid
if (session.kinds_seen() or achievement_records) and not resolved_user_uid:
if (
session.kinds_seen() or achievement_records or gear.snapshot
) and not resolved_user_uid:
resolved_user_uid = console.prompt_user_uid()
if achievement_records:
achievement_path = _achievement_path(resolved_user_uid)
@ -198,6 +206,11 @@ def run_live_capture(
json.dumps(achievement_export, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
gear_path = gear.write(
capture_source=CAPTURE_SOURCE_LABELS.get(capture.name, capture.name),
user_uid=resolved_user_uid,
server_id=session.server_id,
)
resolved_server_id = session.server_id
if session.kinds_seen() and not resolved_server_id:
resolved_server_id = console.prompt_server_id()
@ -251,16 +264,23 @@ def run_live_capture(
console.print_note("No pull history was captured (this is fine if you only wanted achievements).")
print()
console.print_note(f"Export written: {achievement_path}")
elif gear_path is not None:
console.print_success("Gear export complete.")
console.print_note("No pull history was captured (this is fine if you only wanted gear).")
print()
else:
console.print_problem("No history pages were captured.")
console.print_note("Reopen a supported history screen and scroll from page 1.")
console.print_note("If no page messages appear, return to the main menu and re-enter the game.")
if gear_path is not None:
console.print_note(f"Export written: {gear_path}")
if diagnostics_path is not None:
console.print_note(f"Diagnostics written: {diagnostics_path}")
return {
"exports": [],
"diagnostics_path": diagnostics_path,
"achievement_path": achievement_path,
"gear_path": gear_path,
}
for item in exports:
@ -282,6 +302,8 @@ def run_live_capture(
if achievement_path is not None:
print()
console.print_note(f"Export written: {achievement_path}")
if gear_path is not None:
console.print_note(f"Export written: {gear_path}")
if diagnostics_path is not None:
console.print_note(f"Diagnostics written: {diagnostics_path}")
@ -297,6 +319,7 @@ def run_live_capture(
"exports": exports,
"diagnostics_path": diagnostics_path,
"achievement_path": achievement_path,
"gear_path": gear_path,
}

2
tests/test_export_contract.py
View file

@ -8,7 +8,7 @@ from nte_history_exporter.live_capture.runner import _achievement_path
class ExportContractTests(unittest.TestCase):
def test_public_version_references_match(self):
self.assertEqual(__version__, "0.3.0")
self.assertEqual(__version__, "0.4.0")
self.assertEqual(EXPORTER_VERSION, __version__)
def test_sanitized_export_omits_raw_packet_fields(self):

747
tests/test_gear_decoding.py Normal file
View file

@ -0,0 +1,747 @@
"""Gear decoding, driven entirely by synthetic packets.
The wire format is fully understood, so every packet here is constructed from
the spec rather than captured: a 120-bit header, payload bits, two 1 bits and
zero padding. No real capture is used - see tests/fixtures/README.md.
"""
import contextlib
import io
import json
import os
import struct
import tempfile
import unittest
from nte_history_exporter import console
from nte_history_exporter import __version__
from nte_history_exporter.decoder.gear import (
HEADER_BITS,
extract_gear,
payload_end_bit,
splice_datagrams,
)
from nte_history_exporter.export.gear_export import build_gear_export_json
from nte_history_exporter.live_capture import gear_collector, runner
from nte_history_exporter.live_capture.windows_raw import ParsedIpPacket
from tests import support
# --- synthetic wire construction -------------------------------------------
def string_frame(text: bytes) -> bytes:
"""A self-validating <u32 len><ascii\\0> frame."""
return struct.pack("<I", len(text) + 1) + text + b"\0"
def stat_entry(name: bytes, value: float) -> bytes:
"""<u32 len><name\\0><4 zero bytes><f32 value>."""
return string_frame(name) + b"\0\0\0\0" + struct.pack("<f", value)
def poke(buf: bytearray, byte_offset: int, raw: bytes) -> None:
if len(buf) < byte_offset + len(raw):
buf.extend(b"\0" * (byte_offset + len(raw) - len(buf)))
buf[byte_offset : byte_offset + len(raw)] = raw
def poke_u32_at_bit(buf: bytearray, bitpos: int, value: int) -> None:
"""Write a little-endian u32 at an arbitrary bit offset."""
needed = (bitpos + 32 + 7) // 8 + 1
if len(buf) < needed:
buf.extend(b"\0" * (needed - len(buf)))
whole = int.from_bytes(buf, "little")
whole &= ~(0xFFFFFFFF << bitpos)
whole |= value << bitpos
packed = whole.to_bytes(len(buf) + 8, "little")
buf[:] = packed[: len(buf)]
def item_record(
item_id: bytes,
*,
level: int = 20,
entries=(),
instance: bytes = b"\x11\x22\x33\x44\x55\x66\x77\x88",
owner: bytes | None = None,
) -> bytes:
"""One item record laid out at the documented offsets past the id frame."""
body = bytearray(b"\0" * 64)
poke(body, 5, instance)
# Level is a u32 at +38 bytes; the owner reference starts at +39 bits-wise
# (312 bits) and its first three bytes are always zero, so the two fields
# coexist exactly as they do on the wire.
poke(body, 38, struct.pack("<I", level))
if owner is not None:
assert owner[:3] == b"\0\0\0", "owner reference starts with three zero bytes"
poke(body, 39, owner)
chain = b"".join(stat_entry(name, value) for name, value in entries)
return string_frame(item_id) + bytes(body) + chain
def character_block(name: bytes, *, level: int, breakthroughs: int) -> bytes:
"""An ability frame with the progress fields at their negative offsets."""
lead = 128 # bytes of room for the fields that sit before the frame
buf = bytearray(b"\0" * lead)
frame = string_frame(b"GA_" + name + b"_Melee")
block_bit = lead * 8
poke_u32_at_bit(buf, block_bit - 466, 67806)
poke_u32_at_bit(buf, block_bit - 434, level)
poke_u32_at_bit(buf, block_bit - 402, breakthroughs)
return bytes(buf) + frame
def loadout_cell(item_id: bytes, row: int, column: int) -> bytes:
"""An id frame quoted with board coordinates and no stat chain."""
body = bytearray(b"\0" * 32)
# Coordinates sit 96 and 128 bits past the id and are stored doubled.
poke(body, 12, struct.pack("<I", row * 2))
poke(body, 16, struct.pack("<I", column * 2))
return string_frame(item_id) + bytes(body)
def pack_datagrams(payload: bytes, *, chunk_bits: int = 1 << 20, first_seq: int = 1):
"""Split a payload bit stream into framed datagrams."""
total_bits = len(payload) * 8
whole = int.from_bytes(payload, "little")
datagrams, start, seq = [], 0, first_seq
while start < total_bits:
nbits = min(chunk_bits, total_bits - start)
chunk = (whole >> start) & ((1 << nbits) - 1)
header = bytearray(15)
header[0:3] = b"\xaa\xbb\xcc"
# 14-bit packet sequence at bit 24.
header[3] = seq & 0xFF
header[4] = (seq >> 8) & 0x3F
header[12:15] = b"\x64\x06\xee"
# payload bits, then the two 1 bits Unreal terminates with.
value = int.from_bytes(header, "little")
value |= chunk << HEADER_BITS
value |= 0b11 << (HEADER_BITS + nbits)
width = (HEADER_BITS + nbits + 2 + 7) // 8
datagrams.append(value.to_bytes(width, "little"))
start += nbits
seq = (seq + 1) & 0x3FFF
return datagrams
MODULE_ENTRIES = (
(b"HPMaxAdd", 0.0),
(b"AtkAdd", 0.0),
(b"CritDamageBase", 0.06),
(b"CritBase", 0.03),
(b"AtkUp", 0.04),
(b"HPMaxUp", 0.04),
)
CARTRIDGE_ENTRIES = (
(b"AtkUp", 0.0),
(b"CritDamageBase", 0.06),
(b"CritBase", 0.03),
(b"DefUp", 0.07),
(b"HPMaxUp", 0.04),
)
OWNER = b"\0\0\0\x0d\x0d\x20\x3d\x6d\x00\x00\x00\x00"
class FramingTests(unittest.TestCase):
def test_payload_end_bit_finds_the_double_terminator(self):
[datagram] = pack_datagrams(b"\x01\x02\x03\x04")
self.assertEqual(payload_end_bit(datagram), HEADER_BITS + 32)
def test_splice_round_trips_a_payload_across_datagrams(self):
payload = bytes(range(256)) * 3
# A chunk width that is not a multiple of 8 forces the accumulator to
# rejoin the stream at a bit offset, as the real capture does.
datagrams = pack_datagrams(payload, chunk_bits=333)
self.assertGreater(len(datagrams), 1)
self.assertEqual(splice_datagrams(datagrams)[: len(payload)], payload)
def test_pure_acks_are_skipped(self):
payload = b"\x01\x02\x03\x04\x05\x06\x07\x08"
ack = bytes(12)
datagrams = pack_datagrams(payload)
self.assertEqual(splice_datagrams([ack] + datagrams + [ack])[:8], payload)
class ItemRecordTests(unittest.TestCase):
def decode(self, payload: bytes, **kwargs):
return extract_gear(pack_datagrams(b"\0" * 8 + payload, **kwargs))
def test_module_record_decodes(self):
snapshot = self.decode(
item_record(b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES)
)
[item] = snapshot.items
self.assertEqual(item.kind, "module")
self.assertEqual(item.item_id, "cell3_style6_1_Orange")
self.assertEqual(item.shape, "cell3_style6")
self.assertEqual(item.module_type, "III")
self.assertEqual(item.level, 20)
self.assertEqual(item.rarity, "orange")
self.assertEqual(item.instance, "1122334455667788")
self.assertIsNone(item.owner_group)
# Both mains are computed from the cell count, never transmitted.
self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0})
self.assertEqual(len(item.substats), 4)
def test_cartridge_record_decodes_with_its_set(self):
snapshot = self.decode(
item_record(b"Incantation_orange", level=20, entries=CARTRIDGE_ENTRIES)
)
[item] = snapshot.items
self.assertEqual(item.kind, "cartridge")
self.assertIsNotNone(item.set_name)
self.assertEqual(len(item.main_stats), 1)
self.assertEqual(len(item.substats), 4)
def test_owner_reference_is_read_when_equipped(self):
snapshot = self.decode(
item_record(
b"cell3_style6_1_Orange",
level=20,
entries=MODULE_ENTRIES,
owner=OWNER,
)
)
[item] = snapshot.items
self.assertEqual(item.owner_group, OWNER.hex())
# The owner field overlaps the level word without corrupting it.
self.assertEqual(item.level, 20)
def test_record_spanning_a_datagram_boundary_decodes(self):
record = item_record(
b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES, owner=OWNER
)
# Chunks far smaller than the record, at a width that is not a whole
# number of bytes: every field lands across at least one boundary.
snapshot = self.decode(record, chunk_bits=101)
[item] = snapshot.items
self.assertEqual(item.item_id, "cell3_style6_1_Orange")
self.assertEqual(item.level, 20)
self.assertEqual(item.owner_group, OWNER.hex())
self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0})
def test_unframed_id_is_ignored(self):
# Same bytes, wrong length prefix: the frame check must reject it.
record = bytearray(
item_record(b"cell3_style6_1_Orange", entries=MODULE_ENTRIES)
)
record[0:4] = struct.pack("<I", 99)
self.assertEqual(self.decode(bytes(record)).items, ())
class ValidationTests(unittest.TestCase):
def decode_one(self, payload: bytes):
return extract_gear(pack_datagrams(b"\0" * 8 + payload)).items
def test_main_stat_after_a_substat_is_rejected(self):
scrambled = (
(b"HPMaxAdd", 0.0),
(b"CritDamageBase", 0.06),
(b"AtkAdd", 0.0),
(b"CritBase", 0.03),
(b"AtkUp", 0.04),
(b"HPMaxUp", 0.04),
)
self.assertEqual(
self.decode_one(
item_record(b"cell3_style6_1_Orange", entries=scrambled)
),
(),
)
def test_off_table_substat_value_is_rejected(self):
corrupt = MODULE_ENTRIES[:-1] + ((b"HPMaxUp", 0.0413),)
self.assertEqual(
self.decode_one(item_record(b"cell3_style6_1_Orange", entries=corrupt)),
(),
)
def test_short_entry_chain_is_rejected(self):
# Five entries is a cartridge count; a module needs six.
self.assertEqual(
self.decode_one(
item_record(b"cell3_style6_1_Orange", entries=MODULE_ENTRIES[:5])
),
(),
)
class CharacterTests(unittest.TestCase):
def test_level_and_breakthroughs_are_read(self):
payload = character_block(b"Zankou", level=70, breakthroughs=5)
[character] = extract_gear(pack_datagrams(payload)).characters
self.assertEqual(character.key, "Zankou")
self.assertEqual(character.level, 70)
self.assertEqual(character.breakthroughs, 5)
def test_progress_without_the_marker_is_not_reported(self):
payload = bytearray(character_block(b"Zankou", level=70, breakthroughs=5))
poke_u32_at_bit(payload, 128 * 8 - 466, 1) # marker no longer 67806
[character] = extract_gear(pack_datagrams(bytes(payload))).characters
self.assertIsNone(character.level)
self.assertIsNone(character.breakthroughs)
def test_repeated_block_keeps_the_first_valid_reading(self):
block = character_block(b"Zankou", level=70, breakthroughs=5)
blank = bytearray(character_block(b"Zankou", level=70, breakthroughs=5))
poke_u32_at_bit(blank, 128 * 8 - 466, 1)
payload = block + b"\0" * 64 + bytes(blank)
[character] = extract_gear(pack_datagrams(payload)).characters
self.assertEqual(character.level, 70)
self.assertEqual(character.breakthroughs, 5)
class LoadoutTests(unittest.TestCase):
def test_board_cells_follow_their_character_and_are_halved(self):
payload = (
character_block(b"Zankou", level=70, breakthroughs=5)
+ b"\0" * 64
+ loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+ b"\0" * 64
+ loadout_cell(b"cell4_style2_1_Orange", 3, 4)
)
snapshot = extract_gear(pack_datagrams(payload))
[character] = snapshot.characters
# A transposed board renders correctly on a symmetric grid, so the
# assertion has to pin row and column separately.
self.assertEqual(character.board, ((1, 2), (3, 4)))
self.assertEqual(snapshot.characters_with_loadouts, 1)
# Quoted ids are references, not records.
self.assertEqual(snapshot.items, ())
class GearExportContractTests(unittest.TestCase):
def build(self):
payload = (
character_block(b"Zankou", level=70, breakthroughs=5)
+ b"\0" * 64
+ loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+ b"\0" * 64
+ item_record(
b"cell3_style6_1_Orange",
level=20,
entries=MODULE_ENTRIES,
owner=OWNER,
)
+ b"\0" * 64
+ item_record(
b"Incantation_orange",
level=20,
entries=CARTRIDGE_ENTRIES,
instance=b"\x99\xaa\xbb\xcc\xdd\xee\xff\x01",
)
)
snapshot = extract_gear(pack_datagrams(payload))
return snapshot, build_gear_export_json(
snapshot,
source="live_capture",
capture_source="libpcap",
user_uid=" 100200300 ",
server_id="1",
)
def test_envelope(self):
_snapshot, export = self.build()
self.assertEqual(export["format"], "nte-gear-export")
self.assertEqual(export["format_version"], 1)
self.assertEqual(export["source"], "live_capture")
self.assertEqual(export["capture_source"], "libpcap")
self.assertEqual(export["exporter"]["version"], __version__)
self.assertEqual(export["user_uid"], "100200300")
self.assertEqual(export["server_id"], "1")
def test_scan_counts_match_the_snapshot(self):
snapshot, export = self.build()
self.assertEqual(
export["scan"],
{
"cartridges": 1,
"modules": 1,
"characters": 1,
"characters_with_loadouts": 1,
"warnings": [],
},
)
self.assertEqual(len(export["items"]), len(snapshot.items))
def test_item_records_carry_raw_stat_ids(self):
_snapshot, export = self.build()
module = next(i for i in export["items"] if i["kind"] == "module")
self.assertEqual(module["shape"], "cell3_style6")
self.assertEqual(module["module_type"], "III")
self.assertEqual(module["owner_group"], OWNER.hex())
self.assertIn(
{"stat": "HPMaxAdd", "value": 840.0}, module["main_stats"]
)
self.assertTrue(all("stat" in s for s in module["substats"]))
cartridge = next(i for i in export["items"] if i["kind"] == "cartridge")
self.assertIn("set", cartridge)
self.assertIsNone(cartridge["owner_group"])
def test_character_board_and_unresolved_owner_group(self):
_snapshot, export = self.build()
[character] = export["characters"]
self.assertEqual(character["key"], "Zankou")
self.assertEqual(character["level"], 70)
self.assertEqual(character["breakthroughs"], 5)
self.assertEqual(character["board"], [[1, 2]])
# The capture cannot join an owner group to a name; emitting a guess
# would be worse than emitting the truth.
self.assertIsNone(character["owner_group"])
# --- runner wiring ---------------------------------------------------------
class _StubCapture:
name = "libpcap"
detail = "documentation range"
fallback_reason = ""
def __init__(self, packets):
self._packets = packets
def packets(self):
return iter(self._packets)
def stats(self):
return None
def close(self):
return None
LOCAL_IP = "192.0.2.10"
REMOTE_IP = "198.51.100.20"
class RunnerWiringTests(unittest.TestCase):
"""The gear export must survive being the only thing captured."""
def run_capture(self, *, user_uid=None):
payload = (
character_block(b"Zankou", level=70, breakthroughs=5)
+ b"\0" * 64
+ loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+ b"\0" * 64
+ item_record(
b"cell3_style6_1_Orange",
level=20,
entries=MODULE_ENTRIES,
owner=OWNER,
)
)
packets = [
ParsedIpPacket(
src_ip=REMOTE_IP,
dst_ip=LOCAL_IP,
src_port=30166,
dst_port=49610,
payload=datagram,
protocol="udp",
)
for datagram in pack_datagrams(payload, chunk_bits=277)
]
# An outbound datagram and a second, quieter inbound flow: the decoder
# must pick the busiest inbound flow and ignore the rest.
packets.append(
ParsedIpPacket(
src_ip=LOCAL_IP,
dst_ip=REMOTE_IP,
src_port=49610,
dst_port=30166,
payload=b"\x00" * 32,
protocol="udp",
)
)
packets.append(
ParsedIpPacket(
src_ip=REMOTE_IP,
dst_ip=LOCAL_IP,
src_port=30167,
dst_port=49611,
payload=b"\x00" * 16,
protocol="udp",
)
)
prompted = []
def fake_prompt_user_uid():
prompted.append(True)
return "100200300"
with tempfile.TemporaryDirectory() as workdir:
previous = os.getcwd()
os.chdir(workdir)
patched_backend = runner.open_capture_backend
patched_prompt = console.prompt_user_uid
runner.open_capture_backend = lambda ip, backend: _StubCapture(packets)
console.prompt_user_uid = fake_prompt_user_uid
try:
with contextlib.redirect_stdout(io.StringIO()) as out:
result = runner.run_live_capture(
interface_ip=LOCAL_IP, user_uid=user_uid
)
gear_path = result["gear_path"]
self.assertIsNotNone(gear_path)
export = json.loads(gear_path.read_text(encoding="utf-8"))
printed = out.getvalue()
finally:
runner.open_capture_backend = patched_backend
console.prompt_user_uid = patched_prompt
os.chdir(previous)
return result, export, printed, prompted
def test_gear_only_capture_still_writes_and_reports_its_export(self):
result, export, printed, prompted = self.run_capture()
self.assertEqual(result["exports"], [])
self.assertEqual(export["format"], "nte-gear-export")
self.assertEqual(export["scan"]["modules"], 1)
self.assertEqual(export["scan"]["characters_with_loadouts"], 1)
self.assertEqual(export["source"], "live_capture")
self.assertEqual(export["capture_source"], "libpcap")
# The summary branch for "no pull history" has to mention it, or the
# file is written and never named.
self.assertIn("Gear export complete.", printed)
self.assertIn(result["gear_path"].name, printed)
self.assertIn("Gear captured", printed)
# Gear alone must be enough to opt into the UID prompt.
self.assertEqual(prompted, [True])
self.assertEqual(export["user_uid"], "100200300")
self.assertTrue(result["gear_path"].name.startswith("100200300_Gear_"))
def test_supplied_uid_skips_the_prompt(self):
_result, export, _printed, prompted = self.run_capture(user_uid="999")
self.assertEqual(prompted, [])
self.assertEqual(export["user_uid"], "999")
def test_gear_export_is_named_alongside_pull_history_exports(self):
"""The second summary branch: gear must not vanish behind a banner."""
fixture = support.load_network_fixture()
local_ip = fixture["local_ip"]
payload = item_record(
b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES, owner=OWNER
)
packets = [
ParsedIpPacket(
src_ip=packet.src_ip,
dst_ip=packet.dst_ip,
src_port=packet.src_port,
dst_port=packet.dst_port,
payload=packet.payload,
protocol=packet.protocol,
)
for packet in support.fixture_packets("replay")
]
packets += [
ParsedIpPacket(
src_ip=REMOTE_IP,
dst_ip=local_ip,
src_port=30166,
dst_port=49610,
payload=datagram,
protocol="udp",
)
for datagram in pack_datagrams(payload, chunk_bits=277)
]
with tempfile.TemporaryDirectory() as workdir:
previous = os.getcwd()
os.chdir(workdir)
patched_backend = runner.open_capture_backend
patched_server = console.prompt_server_id
runner.open_capture_backend = lambda ip, backend: _StubCapture(packets)
console.prompt_server_id = lambda: ""
try:
with contextlib.redirect_stdout(io.StringIO()) as out:
result = runner.run_live_capture(
interface_ip=local_ip, user_uid="100200300"
)
printed = out.getvalue()
gear_path = result["gear_path"]
self.assertIsNotNone(gear_path)
export = json.loads(gear_path.read_text(encoding="utf-8"))
finally:
runner.open_capture_backend = patched_backend
console.prompt_server_id = patched_server
os.chdir(previous)
self.assertTrue(result["exports"], "the fixture should still yield a banner")
self.assertEqual(export["scan"]["modules"], 1)
self.assertIn(gear_path.name, printed)
def test_udp_flow_accumulation_is_bounded(self):
self.assertEqual(gear_collector.MAX_UDP_FLOW_BYTES, 8 * 1024 * 1024)
class _FakeClock:
"""Stands in for the `time` module so idleness is exact, not slept for."""
def __init__(self, now=1000.0):
self.now = now
def monotonic(self):
return self.now
def time(self):
return self.now
class _TickingCapture(_StubCapture):
"""Yields a burst, then the read-timeout ticks pcap emits when idle.
`pending` is how many items are still unyielded, which is what proves a
report happened during the capture rather than after it.
"""
def __init__(self, packets, clock, *, ticks=3, tick_seconds=5.0):
super().__init__(packets)
self.clock = clock
self.ticks = ticks
self.tick_seconds = tick_seconds
self.pending = 0
def packets(self):
queued = list(self._packets) + [None] * self.ticks
self.pending = len(queued)
for item in queued:
self.pending -= 1
if item is None:
self.clock.now += self.tick_seconds
yield item
class LiveGearReportingTests(unittest.TestCase):
"""Gear is reported while the capture runs, not only once it stops."""
def burst_packets(self):
payload = (
character_block(b"Zankou", level=70, breakthroughs=5)
+ b"\0" * 64
+ loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+ b"\0" * 64
+ item_record(
b"cell3_style6_1_Orange",
level=20,
entries=MODULE_ENTRIES,
owner=OWNER,
)
)
return [
ParsedIpPacket(
src_ip=REMOTE_IP,
dst_ip=LOCAL_IP,
src_port=30166,
dst_port=49610,
payload=datagram,
protocol="udp",
)
for datagram in pack_datagrams(payload, chunk_bits=277)
]
def run_capture(self, capture_factory, *, min_flow_bytes=1):
"""Run one live capture against a stub, with the size gate relaxed.
Synthetic bursts are a few hundred bytes; the real gate exists to keep
chatter out, not to describe a minimum decodable size.
"""
clock = _FakeClock()
capture = capture_factory(clock)
decodes = []
real_extract = gear_collector.extract_gear
reported_pending = []
real_report = gear_collector.print_gear_captured
def counting_extract(grams):
decodes.append(len(grams))
return real_extract(grams)
def recording_report(*args, **kwargs):
reported_pending.append(capture.pending)
return real_report(*args, **kwargs)
with tempfile.TemporaryDirectory() as workdir:
previous = os.getcwd()
os.chdir(workdir)
patched_backend = runner.open_capture_backend
patched_time = gear_collector.time
runner.open_capture_backend = lambda ip, backend: capture
gear_collector.time = clock
gear_collector.extract_gear = counting_extract
gear_collector.print_gear_captured = recording_report
previous_min = gear_collector.GEAR_LIVE_MIN_FLOW_BYTES
gear_collector.GEAR_LIVE_MIN_FLOW_BYTES = min_flow_bytes
try:
with contextlib.redirect_stdout(io.StringIO()) as out:
result = runner.run_live_capture(
interface_ip=LOCAL_IP, user_uid="100200300"
)
printed = out.getvalue()
finally:
runner.open_capture_backend = patched_backend
gear_collector.time = patched_time
gear_collector.extract_gear = real_extract
gear_collector.print_gear_captured = real_report
gear_collector.GEAR_LIVE_MIN_FLOW_BYTES = previous_min
os.chdir(previous)
return result, printed, decodes, reported_pending
def test_gear_is_reported_before_the_capture_ends(self):
result, printed, _, reported_pending = self.run_capture(
lambda clock: _TickingCapture(self.burst_packets(), clock)
)
self.assertIn("Gear captured", printed)
self.assertEqual(len(reported_pending), 1, "reported exactly once")
self.assertGreater(
reported_pending[0],
0,
"the report must land while packets are still to come",
)
self.assertIsNotNone(result["gear_path"])
def test_a_flow_that_never_goes_idle_still_decodes_at_the_stop(self):
result, printed, _, reported_pending = self.run_capture(
lambda clock: _TickingCapture(self.burst_packets(), clock, ticks=0)
)
self.assertIn("Gear captured", printed)
self.assertEqual(
reported_pending, [0], "nothing pending: this is the post-loop fallback"
)
self.assertIsNotNone(result["gear_path"])
def test_an_idle_flow_is_not_decoded_twice_for_the_same_bytes(self):
# 40 ticks against a flow that never yields gear: without the
# already-attempted guard this would decode once per tick.
junk = [
ParsedIpPacket(
src_ip=REMOTE_IP,
dst_ip=LOCAL_IP,
src_port=30166,
dst_port=49610,
payload=b"\x00" * 64,
protocol="udp",
)
]
_, _, decodes, _ = self.run_capture(
lambda clock: _TickingCapture(junk, clock, ticks=40)
)
# One live attempt for the single flow size, plus the post-loop scan.
self.assertEqual(len(decodes), 2)
if __name__ == "__main__":
unittest.main()