hashbro
150 lines
5.0 KiB
Python
150 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
wap.js payload -> __ws.txt NDJSON (chk.ts native output format)
|
|
|
|
Route-1: infer cc/country from phone via libphonenumber, derive `in` by
|
|
stripping cc from phone, derive clientStaticPublicKey from the private key
|
|
via curve25519.
|
|
|
|
Usage:
|
|
python wapjs_to_ws.py <input.json> [output.ndjson]
|
|
|
|
If output omitted, writes <input-stem>.ndjson next to input.
|
|
"""
|
|
import sys, json, base64, phonenumbers
|
|
from pathlib import Path
|
|
from nacl.public import PrivateKey # curve25519
|
|
|
|
|
|
# ---------- protobuf (minimal, only what chk.ts signedPreKey needs) ----------
|
|
def _varint(d, o):
|
|
v = s = 0
|
|
while True:
|
|
b = d[o]; o += 1
|
|
v |= (b & 0x7f) << s
|
|
if not (b & 0x80): break
|
|
s += 7
|
|
return v, o
|
|
|
|
def parse_pb(d):
|
|
out, o = {}, 0
|
|
while o < len(d):
|
|
tag, o = _varint(d, o)
|
|
fn, w = tag >> 3, tag & 7
|
|
if w == 0:
|
|
v, o = _varint(d, o)
|
|
elif w == 1:
|
|
v = d[o:o + 8]; o += 8
|
|
elif w == 2:
|
|
ln, o = _varint(d, o)
|
|
v = d[o:o + ln]; o += ln
|
|
elif w == 5:
|
|
v = d[o:o + 4]; o += 4
|
|
else:
|
|
raise ValueError(f"bad wire {w} field {fn}")
|
|
out[fn] = v
|
|
return out
|
|
|
|
|
|
def b64(b: bytes) -> str:
|
|
return base64.b64encode(b).decode()
|
|
|
|
|
|
def infer_cc_country(phone_int: int):
|
|
"""Return (cc, country_iso, in_local) using libphonenumber."""
|
|
s = "+" + str(phone_int)
|
|
try:
|
|
nn = phonenumbers.parse(s, None)
|
|
if not phonenumbers.is_valid_number(nn):
|
|
# still try to get region from prefix even if invalid
|
|
region = phonenumbers.region_code_for_country_code(nn.country_code) or ""
|
|
else:
|
|
region = phonenumbers.region_code_for_number(nn) or ""
|
|
cc = str(nn.country_code)
|
|
national = str(nn.national_number)
|
|
return cc, region, national
|
|
except phonenumbers.NumberParseException:
|
|
return "", "", str(phone_int)
|
|
|
|
|
|
def convert(wap_path: Path) -> str:
|
|
o = json.loads(wap_path.read_text())
|
|
pks = o["phoneKeyStore"]
|
|
ident = pks["identity"]
|
|
spk = parse_pb(bytes.fromhex(pks["signedPreKey"]["hexKey"]))
|
|
dc = o.get("deviceConfig", {})
|
|
|
|
cc, country, in_local = infer_cc_country(int(o["userId"]))
|
|
phone = str(o["userId"])
|
|
|
|
# identity keys (keep 05 prefix)
|
|
ident_pub_b = bytes.fromhex(ident["hexPublic"]) # 33 bytes
|
|
ident_priv_b = bytes.fromhex(ident["hexPrivate"]) # 32 bytes
|
|
|
|
# signed prekey (protobuf): 1=id 2=pub(33,05+) 3=priv(32) 4=sig(64)
|
|
spk_id = spk[1]
|
|
spk_pub_b = spk[2] # 33 bytes
|
|
spk_priv_b = spk[3] # 32 bytes
|
|
spk_sig_b = spk[4] # 64 bytes
|
|
|
|
# clientStatic: private given, derive public (raw 32 bytes, no 05 prefix)
|
|
cs_priv_b = base64.b64decode(o["clientStaticKeypairBase64"])
|
|
cs_pub_b = bytes(PrivateKey(cs_priv_b).public_key) # 32 bytes
|
|
|
|
record = {
|
|
"cc": cc,
|
|
"clientStaticPrivateKey": b64(cs_priv_b),
|
|
"clientStaticPublicKey": b64(cs_pub_b),
|
|
"country": country,
|
|
"device": dc.get("model", ""),
|
|
"deviceUUID": "",
|
|
"identityPrivateKey": b64(ident_priv_b),
|
|
"identityPublicKey": b64(ident_pub_b),
|
|
"in": in_local,
|
|
"jid": phone,
|
|
"language": "",
|
|
"manufacturer": dc.get("brand", "Apple") or "Apple",
|
|
"mcc": dc.get("sim_operator", ""),
|
|
"mnc": "",
|
|
"osBuildNumber": dc.get("display", ""),
|
|
"osVersion": dc.get("sdk_release", ""),
|
|
"phone": phone,
|
|
"phoneUUID": o.get("phoneId", ""),
|
|
"registrationID": ident.get("registration_id", 0),
|
|
"roProductBoard": dc.get("board", ""),
|
|
"roProductDevice": dc.get("device", ""),
|
|
"signPreKeyID": spk_id,
|
|
"signPreKeyPrivateKey": b64(spk_priv_b),
|
|
"signPreKeyPublicKey": b64(spk_pub_b),
|
|
"signPreKeySignature": b64(spk_sig_b),
|
|
"whatsappVersion": "",
|
|
}
|
|
return json.dumps(record, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print(__doc__); sys.exit(1)
|
|
inp = Path(sys.argv[1])
|
|
out = Path(sys.argv[2]) if len(sys.argv) > 2 else inp.with_suffix(".ndjson")
|
|
line = convert(inp)
|
|
out.write_text(line + "\n")
|
|
print(f"wrote {out} ({len(line)} chars)")
|
|
# echo parsed summary
|
|
r = json.loads(line)
|
|
print("\n=== summary ===")
|
|
for k in ["cc","country","in","phone","jid","phoneUUID","registrationID",
|
|
"device","roProductDevice","roProductBoard","osVersion","osBuildNumber",
|
|
"manufacturer","mcc"]:
|
|
print(f" {k:18s} = {r[k]!r}")
|
|
print(" --- key lengths (raw bytes) ---")
|
|
for k in ["identityPublicKey","identityPrivateKey","signPreKeyPublicKey",
|
|
"signPreKeyPrivateKey","signPreKeySignature",
|
|
"clientStaticPrivateKey","clientStaticPublicKey"]:
|
|
b = base64.b64decode(r[k])
|
|
print(f" {k:22s} = {len(b):3d} bytes head={b[:3].hex()}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|