"""Read-only archive inventory and schema extraction; never exports credentials or replays requests."""
import argparse
import collections
import hashlib
import json
import pathlib
import urllib.parse
import zipfile
parser = argparse.ArgumentParser()
parser.add_argument("archive", type=pathlib.Path)
parser.add_argument("--output", type=pathlib.Path, required=True)
parser.add_argument("--fixture", type=pathlib.Path)
args = parser.parse_args()
archive_hash = hashlib.file_digest(args.archive.open("rb"), "sha256").hexdigest()
records = {}
market_fixtures = []
with zipfile.ZipFile(args.archive) as archive:
entries = archive.infolist()
for entry in entries:
if not entry.filename.endswith("/events.jsonl"):
continue
folder = entry.filename.removesuffix("events.jsonl")
for line in archive.read(entry).decode("utf-8").splitlines():
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
if event.get("type") != "http-response":
continue
path = str(event.get("path", "")).split("?")[0]
key = (event.get("host", ""), path, event.get("method", ""))
record = records.setdefault(key, {"host": key[0], "path": path, "method": key[2], "count": 0, "statuses": {}, "queryFields": [], "responseFields": [], "source": entry.filename})
record["count"] += 1
status = str(event.get("status", ""))
record["statuses"][status] = record["statuses"].get(status, 0) + 1
query = urllib.parse.parse_qs(str(event.get("query", "")))
query_name = folder + f"decoded/http-response-{int(event.get('id', 0)):06}.query.json"
try:
decoded_query = json.loads(archive.read(query_name))
if isinstance(decoded_query, list):
query = {x[0]: True for x in decoded_query if isinstance(x, list) and len(x) == 2}
elif isinstance(decoded_query, dict):
query = {k: True for k in decoded_query}
except (KeyError, ValueError):
pass
record["queryFields"] = sorted(set(record["queryFields"]) | set(query))
body_name = folder + f"decoded/http-response-{int(event.get('id', 0)):06}.json"
try:
body = json.loads(archive.read(body_name))
except (KeyError, ValueError):
continue
if not isinstance(body, dict):
continue
record["responseFields"] = sorted(set(record["responseFields"]) | set(body))
if path in ("/2.0/elite/market/list", "/2.0/elite/market/trade") and "commodities" in body:
commodities = body["commodities"]
items = list(commodities.values()) if isinstance(commodities, dict) else commodities
record["commodityCount"] = len(items)
record["commodityFields"] = sorted(set().union(*(set(x) for x in items)))
record["inventoryFields"] = sorted(set().union(*(set(x) for x in body.get("inventory", []))))
# Strict allowlist: commodity-only fixture. Account state and inventory are discarded.
permitted = {"id", "categoryname", "name", "stock", "buyPrice", "sellPrice", "demand", "meanPrice", "legality", "rare"}
market_fixtures.append({"sourceEntry": body_name, "observedAt": event.get("utc"), "commodities": [{k: v for k, v in x.items() if k in permitted} for x in items[:4]]})
report = {"archiveName": args.archive.name, "sha256": archive_hash, "entries": len(entries), "uncompressedBytes": sum(x.file_size for x in entries), "pcapCount": sum(x.filename.endswith(".pcapng") for x in entries), "httpEndpoints": list(records.values()), "scope": "Observed HTTP schemas only. No request replay, no credential export, no game automation validation.", "clientValidated": False}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
if args.fixture and market_fixtures:
args.fixture.parent.mkdir(parents=True, exist_ok=True)
args.fixture.write_text(json.dumps(market_fixtures, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps({"sha256": archive_hash, "entries": report["entries"], "observedHttpEndpoints": len(records), "marketResponses": len(market_fixtures), "report": str(args.output)}, ensure_ascii=False))
"""Read-only archive inventory and schema extraction; never exports credentials or replays requests."""
import argparse
import collections
import hashlib
import json
import pathlib
import urllib.parse
import zipfile
parser = argparse.ArgumentParser()
parser.add_argument("archive", type=pathlib.Path)
parser.add_argument("--output", type=pathlib.Path, required=True)
parser.add_argument("--fixture", type=pathlib.Path)
args = parser.parse_args()
archive_hash = hashlib.file_digest(args.archive.open("rb"), "sha256").hexdigest()
records = {}
market_fixtures = []
with zipfile.ZipFile(args.archive) as archive:
entries = archive.infolist()
for entry in entries:
if not entry.filename.endswith("/events.jsonl"):
continue
folder = entry.filename.removesuffix("events.jsonl")
for line in archive.read(entry).decode("utf-8").splitlines():
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
if event.get("type") != "http-response":
continue
path = str(event.get("path", "")).split("?")[0]
key = (event.get("host", ""), path, event.get("method", ""))
record = records.setdefault(key, {"host": key[0], "path": path, "method": key[2], "count": 0, "statuses": {}, "queryFields": [], "responseFields": [], "source": entry.filename})
record["count"] += 1
status = str(event.get("status", ""))
record["statuses"][status] = record["statuses"].get(status, 0) + 1
query = urllib.parse.parse_qs(str(event.get("query", "")))
query_name = folder + f"decoded/http-response-{int(event.get('id', 0)):06}.query.json"
try:
decoded_query = json.loads(archive.read(query_name))
if isinstance(decoded_query, list):
query = {x[0]: True for x in decoded_query if isinstance(x, list) and len(x) == 2}
elif isinstance(decoded_query, dict):
query = {k: True for k in decoded_query}
except (KeyError, ValueError):
pass
record["queryFields"] = sorted(set(record["queryFields"]) | set(query))
body_name = folder + f"decoded/http-response-{int(event.get('id', 0)):06}.json"
try:
body = json.loads(archive.read(body_name))
except (KeyError, ValueError):
continue
if not isinstance(body, dict):
continue
record["responseFields"] = sorted(set(record["responseFields"]) | set(body))
if path in ("/2.0/elite/market/list", "/2.0/elite/market/trade") and "commodities" in body:
commodities = body["commodities"]
items = list(commodities.values()) if isinstance(commodities, dict) else commodities
record["commodityCount"] = len(items)
record["commodityFields"] = sorted(set().union(*(set(x) for x in items)))
record["inventoryFields"] = sorted(set().union(*(set(x) for x in body.get("inventory", []))))
# Strict allowlist: commodity-only fixture. Account state and inventory are discarded.
permitted = {"id", "categoryname", "name", "stock", "buyPrice", "sellPrice", "demand", "meanPrice", "legality", "rare"}
market_fixtures.append({"sourceEntry": body_name, "observedAt": event.get("utc"), "commodities": [{k: v for k, v in x.items() if k in permitted} for x in items[:4]]})
report = {"archiveName": args.archive.name, "sha256": archive_hash, "entries": len(entries), "uncompressedBytes": sum(x.file_size for x in entries), "pcapCount": sum(x.filename.endswith(".pcapng") for x in entries), "httpEndpoints": list(records.values()), "scope": "Observed HTTP schemas only. No request replay, no credential export, no game automation validation.", "clientValidated": False}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
if args.fixture and market_fixtures:
args.fixture.parent.mkdir(parents=True, exist_ok=True)
args.fixture.write_text(json.dumps(market_fixtures, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps({"sha256": archive_hash, "entries": report["entries"], "observedHttpEndpoints": len(records), "marketResponses": len(market_fixtures), "report": str(args.output)}, ensure_ascii=False))
