172 lines
5.0 KiB
Python
172 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
decrypt_capture.py - Offline decryption of captured proxy traffic.
|
|
Reads a proxy_*.jsonl capture file and tries to decrypt with a given key.
|
|
Useful for testing keys against captured data.
|
|
|
|
Usage:
|
|
python decrypt_capture.py [capture_file] [key_file]
|
|
Defaults: latest proxy_*.jsonl and captures/chacha_session_key.json
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import json
|
|
import glob
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from Crypto.Cipher import ChaCha20
|
|
import paralogue_net_full_pb2 as pb
|
|
|
|
CAPTURES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'captures')
|
|
|
|
|
|
def find_latest_capture():
|
|
files = glob.glob(os.path.join(CAPTURES_DIR, 'proxy_*.jsonl'))
|
|
if not files:
|
|
files = glob.glob(os.path.join(CAPTURES_DIR, 'decproxy_*.jsonl'))
|
|
if not files:
|
|
return None
|
|
return max(files, key=os.path.getmtime)
|
|
|
|
|
|
def load_key(key_file):
|
|
with open(key_file, 'r') as f:
|
|
data = json.load(f)
|
|
if isinstance(data, list):
|
|
data = data[0]
|
|
key = bytes.fromhex(data['key_hex'])
|
|
nonce = bytes.fromhex(data['nonce_hex'])
|
|
return key, nonce
|
|
|
|
|
|
def try_parse_packet_head(data):
|
|
"""Try to parse decrypted data as PacketHead."""
|
|
try:
|
|
ph = pb.PacketHead()
|
|
ph.ParseFromString(data)
|
|
if ph.msg_id > 0 and ph.body_len >= 0:
|
|
return ph
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
def try_decrypt_stream(packets, key, nonce):
|
|
"""Try decrypting a sequence of packets as a continuous stream."""
|
|
c2s_cipher = ChaCha20.new(key=key, nonce=nonce)
|
|
s2c_cipher = ChaCha20.new(key=key, nonce=nonce)
|
|
|
|
results = []
|
|
for pkt in packets:
|
|
direction = pkt['d']
|
|
raw = bytes.fromhex(pkt['hex'])
|
|
cipher = c2s_cipher if direction == 'C->S' else s2c_cipher
|
|
|
|
dec = cipher.encrypt(raw) # XOR = same for enc/dec
|
|
ph = try_parse_packet_head(dec)
|
|
|
|
result = {
|
|
'n': pkt['n'],
|
|
'd': direction,
|
|
'sz': pkt['sz'],
|
|
'dec_hex': dec[:64].hex(),
|
|
}
|
|
|
|
if ph and ph.msg_id > 0:
|
|
result['packet_head'] = {
|
|
'packet_id': ph.packet_id,
|
|
'msg_id': ph.msg_id,
|
|
'flag': ph.flag,
|
|
'body_len': ph.body_len,
|
|
'total_pack_count': ph.total_pack_count,
|
|
'seq_id': ph.seq_id,
|
|
}
|
|
# Try to parse body
|
|
head_size = ph.ByteSize()
|
|
body = dec[head_size:head_size + ph.body_len]
|
|
if body:
|
|
result['body_hex'] = body[:64].hex()
|
|
result['body_len_actual'] = len(body)
|
|
|
|
results.append(result)
|
|
return results
|
|
|
|
|
|
def main():
|
|
cap_file = sys.argv[1] if len(sys.argv) > 1 else find_latest_capture()
|
|
key_file = sys.argv[2] if len(sys.argv) > 2 else os.path.join(
|
|
CAPTURES_DIR, 'chacha_session_key.json')
|
|
|
|
if not cap_file:
|
|
print('[!] No capture file found')
|
|
sys.exit(1)
|
|
if not os.path.exists(key_file):
|
|
print(f'[!] Key file not found: {key_file}')
|
|
sys.exit(1)
|
|
|
|
print(f'Capture: {cap_file}')
|
|
print(f'Key: {key_file}')
|
|
|
|
key, nonce = load_key(key_file)
|
|
print(f'Key: {key.hex()}')
|
|
print(f'Nonce: {nonce.hex()} ({len(nonce)} bytes)')
|
|
print()
|
|
|
|
# Load packets, skip handshake (KEY_A/KEY_B)
|
|
packets = []
|
|
with open(cap_file, 'r') as f:
|
|
for line in f:
|
|
pkt = json.loads(line.strip())
|
|
label = pkt.get('label', '')
|
|
# Skip handshake packets
|
|
if 'KEY_A' in label or 'KEY_B' in label or 'type_100' in label:
|
|
print(f' Skip handshake: #{pkt["n"]} {label}')
|
|
continue
|
|
packets.append(pkt)
|
|
|
|
print(f'\n{len(packets)} encrypted packets to decrypt\n')
|
|
|
|
# Try with 12-byte nonce
|
|
results = try_decrypt_stream(packets, key, nonce)
|
|
|
|
# Check if decryption looks valid
|
|
valid = 0
|
|
for r in results:
|
|
if 'packet_head' in r:
|
|
valid += 1
|
|
|
|
print(f'Valid PacketHead parses: {valid}/{len(results)}')
|
|
print()
|
|
|
|
if valid == 0 and len(nonce) == 12:
|
|
# Try 8-byte nonce
|
|
print('Trying 8-byte nonce...')
|
|
results = try_decrypt_stream(packets, key, nonce[:8])
|
|
valid = sum(1 for r in results if 'packet_head' in r)
|
|
print(f'Valid with 8B nonce: {valid}/{len(results)}')
|
|
|
|
# Print results
|
|
for r in results[:20]:
|
|
ph = r.get('packet_head')
|
|
if ph:
|
|
print(f' #{r["n"]:04d} {r["d"]} {r["sz"]:5d}B -> '
|
|
f'msg_id={ph["msg_id"]} pkt={ph["packet_id"]} '
|
|
f'body={ph["body_len"]}B flag={ph["flag"]}')
|
|
else:
|
|
print(f' #{r["n"]:04d} {r["d"]} {r["sz"]:5d}B -> '
|
|
f'[no valid header] {r["dec_hex"][:32]}...')
|
|
|
|
if len(results) > 20:
|
|
print(f' ... and {len(results)-20} more')
|
|
|
|
# Save full results
|
|
out_file = cap_file.replace('.jsonl', '_decrypted.json')
|
|
with open(out_file, 'w') as f:
|
|
json.dump(results, f, indent=2)
|
|
print(f'\nFull results saved to {out_file}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|