CHIRP

#! /usr/bin/python

# Test script to play with the Wouxun KG-UV9D-Plus

# Based on the notes and work from the the Wouxun KG-UV8D Plus

# Pavel Milanes, CO7WT, pavelmc@gmail.com

# some notes from the chirp's driver

# Figured out how the data is encrypted and implement

# serial data encryption and decryption functions.

# The algorithm of decryption works like this:

# - the first byte of data stream is XOR by const 57h

# - each next byte is encoded by previous byte using the XOR

# including the checksum (e.g data[i - 1] xor data[i])

# Support for the Wouxun KG-UV8D Plus radio

# Serial coms are at 19200 baud

# The data is passed in variable length records

# Record structure:

# Offset Usage

# 0 start of record (\x7a)

# 1 Command (\x80 Identify \x81 End/Reboot \x82 Read \x83 Write)

# 2 direction (\xff PC-> Radio, \x00 Radio -> PC)

# 3 length of payload (excluding header/checksum) (n)

# 4 payload (n bytes)

# 4+n+1 checksum - byte sum (% 256) of bytes 1 -> 4+n

#

# Memory Read Records:

# the payload is 3 bytes, first 2 are offset (big endian),

# 3rd is number of bytes to read

# Memory Write Records:

# the maximum payload size (from the Wouxun software) seems to be 66 bytes

# (2 bytes location + 64 bytes data).

import sys

import struct

import string

cmd_name = {

0x80 : "ident",

0x81 : "hangup",

0x82 : "read config",

0x83 : "write config",

0x84 : "read channel memory",

0x85 : "write channel memory"}

# list of known config items. If it isn't here, we don't know it yet

config_item_type = {

0x740: "FM chan 1-8",

0x750: "FM chan 9-16",

0x760: "FM chan 17-20",

0x8b0: "Configuration",

0x8c0: "PTT-ANI",

0x8d0: "SCC",

0x8f0: "Display banner",

0x940: "Scan groups part1",

0x980: "Scan groups part2",

0x7400: "CALL-ID 1-8",

0x7440: "CALL-ID 9-16",

0x7480: "CALL-ID 17-20, names 0-4",

0x74c0: "CALL-ID names 5-12",

0x7500: "CALL-ID names 13-20"

}

# error counters

cmd_cnt = {}

rep_cnt = {}

cmd_cerrs = {}

rep_cerrs = {}

cmd_lerrs = {}

rep_lerrs = {}

def hexprint(data, addrfmt=None):

"""Return a hexdump-like encoding of @data"""

if addrfmt is None:

addrfmt = '%(addr)03i'

block_size = 16

lines = (len(data) / block_size)

if (len(data) % block_size > 0):

lines += 1

out = ""

left = len(data)

for block in range(0, lines):

addr = block * block_size

try:

out += addrfmt % locals()

except (OverflowError, ValueError, TypeError, KeyError):

out += "%03i" % addr

out += ': '

if left < block_size:

limit = left

else:

limit = block_size

for j in range(0, block_size):

if (j < limit):

out += "%02x " % data[(block * block_size) + j]

else:

out += " "

out += " "

for j in range(0, block_size):

if (j < limit):

_byte = data[(block * block_size) + j]

if _byte >= 0x20 and _byte < 0x7F:

out += "%s" % chr(_byte)

else:

out += "."

else:

out += " "

out += "\n"

if (left > block_size):

left -= block_size

return out

def bcd_encode(val, bigendian=True, width=None):

"""This is really old and shouldn't be used anymore"""

digits = []

while val != 0:

digits.append(val % 10)

val /= 10

result = ""

if len(digits) % 2 != 0:

digits.append(0)

while width and width > len(digits):

digits.append(0)

for i in range(0, len(digits), 2):

newval = struct.pack("B", (digits[i + 1] << 4) | digits[i])

if bigendian:

result = newval + result

else:

result = result + newval

return result

def pkt_encode(op, payload):

"""Assemble a packet for the radio and encode it for transmission.

Yes indeed, the checksum we store is only 4 bits. Why, I suspect it's a bug in

the radio firmware guys didn't want to fix, i.e. a typo 0xff -> 0xf..."""

data = bytearray()

data.append(0x7d) # tag that marks the beginning of the packet

data.append(op)

data.append(0xff) # 0xff is from app to radio

data.append(len(payload))

# calc checksum from op to end

cksum = op + 0xff + len(payload)

for byte in payload:

cksum += byte

data.append(byte)

data.append(cksum & 0xf) # Yea, this is a 4 bit cksum (also known as a bug)

# now obfuscate by an xor starting with first payload byte ^ 0x52 including the trailing cksum.

xorbits = 0x52

for i, byte in enumerate(data[4:]):

xord = xorbits ^ byte

data[i + 4] = xord

xorbits = xord

return(data)

def pkt_decode(data):

"""Take a packet hot off the wire and decode it into clear text and return the fields.

We say <<cleartext>> here because all it turns out to be is annoying obfuscation.

This is the inverse of pkt_decode"""

errs = ""

# we don't care about data[0]. It is always 0x7d sl leave it alone

op = data[1]

direction = data[2]

bytecount = data[3]

# First un-obfuscate the payload and cksum

payload = bytearray()

xorbits = 0x52

for i, byte in enumerate(data[4:]):

payload.append(xorbits ^ byte)

xorbits = byte

# Calculate the checksum starting with the 3 bytes of the header

cksum = op + direction + bytecount

for byte in payload[:-1]:

cksum += byte

cksum_match = (cksum & 0xf) == payload[-1] # yes, a 4 bit cksum to match the encode

if (not cksum_match):

errs += "Checksum missmatch: %x != %x; " % (cksum, payload[-1])

return (cksum_match, op, direction, bytecount, payload[:-1], errs)

# display helpers

def short2freq(num):

""" Convert a signed 16 bit integer into a frequency string expressed in MHz

This is stored in the radio as units of 10KHz which we compensate to Hz.

A value of -1 indicates <no freqency>, i.e. unused channel.

"""

if (num > 0):

return "%2d.%01d" % (num/10, num % 10)

else:

return "----"

def int2freq(num):

""" Convert a signed 32 bit integer into a frequency string expressed in MHz

This is stored in the radio as units of 10Hz which we compensate to Hz.

A value of -1 indicates <no freqency>, i.e. unused channel.

"""

if (num > 0):

return "%3d.%05d" % (num/100000, num % 100000)

else:

return "--none---"

def callid2str(blk):

"""Caller ID per MDC-1200 spec? Must be 3-6 digits (100 - 999999).

Note that the supplied software has a bug that allows '#' and an LF (0x0a)

to be inserted. We do not check this (yet)

"""

bin2ascii = string.maketrans("\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09",

"0123456789")

cid = struct.unpack(">6s2x", blk)

return cid[0].translate(bin2ascii)

def str2callid(s):

""" See callid2str. This must edit check for numbers and range

"""

blk = bytearray()

for c in s:

blk.append(ord(c))

blk.append(0x00)

blk.append(0x00)

return blk

def conf_item_name(item):

""" Return a name for a config item/addres if we know it

"""

if (item in config_item_type):

name = config_item_type[item]

else:

name = "0x%x" % item

return name

# config content encoder/decoders

# 0x740, 0x750, 0x760

def display_fm_chans(addr, blk):

""" Display FM band channels. These are stored as 16 bit signed integers in units

of 10KHz. They seem to be read in 32 byte chunks and written in 8 byte chunks.

This is a simple structure.

struct fm_chan {

int16_t rx;

} chan[n]; /* where n is a multiple of 8 in range 8-48 although the read returns 40

"""

chan = ((addr - 0x740)>>1) + 1

while (len(blk) > 0):

f = struct.unpack('>h', blk[0:2])

print("chan %2d: %s" % (chan, short2freq(f[0])))

blk = blk[2:]

chan += 1

# 0x8b0

def display_radio_conf(addr, blk):

"""General radio configuration controls

"""

cnames = {

0: "c0",

1: "c1",

2: "c2",

3: "c3",

4: "c4",

5: "c5",

6: "c6",

7: "c7",

8: "c8",

9: "c9",

10: "c10",

11: "Power on display",

12: "c12",

13: "c13",

14: "c14",

15: "c15"

}

for i, name in enumerate(blk):

if (i == 11):

if (blk[i] == 0x00):

val = "BATTERY"

else:

val = "BITMAP"

else:

val = "0x%02x" % blk[i]

print("%s: %s" % (cnames[i], val))

# 0x8c0

def display_ptt_id(addr, blk):

"""Radio sends its caller id on transmission

"""

cid = callid2str(blk[0:8])

bits = blk[8:]

print("My call ID: %s, bits %s" % (cid, hexprint(bits)))

# 0x8d0

def display_scc_control(addr, blk):

"""This is the stun/kill/revive code and control?

This is not definitive yet. It doesn't brick the radio but still...

"""

scc = callid2str(blk[0:8])

bits = blk[8:]

print("My SCC code: %s, bits %s" % (scc, hexprint(bits)))

# 0x940, 0x980

group_chan = {}

group_names = {}

def display_scan_group(item, blk):

""" Display the scan group list. It spans two content blocks

block 0x940 is the first part followed by block 0x980 which

contains the remaining names. In real life, these config read/write

could come in any order. for here, we assume this order.

struct scan_940 {

struct {

int16_t start_chan;

int16_t end_chan;

} groups[10];

unsigned char[8] padding;

struct {

char name[8];

} names[2];

}

struct scan_980 {

struct {

char name[8];

} names[8]

}

"""

global group_chan

global group_names

if (item == 0x940):

# first pick up the start end pairs for each group

group_chan = {}

group_names = {} # first block so clear anything old/stale

for i in range(0,10):

rec = blk[0:4]

(start, end) = struct.unpack('>hh', rec)

group_chan[i] = (start, end)

blk = blk[4:]

blk = blk[8:] # align over padding

# Followed by the first two group names

for i in range(0,2):

rec = blk[0:8]

name = struct.unpack('8s', rec)

group_names[i] = name[0].replace('\0', ' ') # make trailing nulls into spaces for pretty

blk = blk[8:]

# Put it together for the first two groups.

print("Address: 0x%x" % item)

for i in range(0,2):

start, end = group_chan[i]

print("group %2d '%s': channel %i to %i" % (i+1, group_names[i], start, end))

else:

# We unpack and display the remaining groups assuming 0x940 came first...

for i in range(2,10):

rec = blk[0:8]

name = struct.unpack('8s', rec)

group_names[i] = name[0].replace('\0', ' ') # make trailing nulls into spaces for pretty

blk = blk[8:]

# Put it together for the first two groups.

print("Address: 0x%x" % item)

for i in range(2,10):

start, end = group_chan[i]

print("group %2d '%s': channel %i to %i" % (i+1, group_names[i], start, end))

# 0x7400, 0x7440, 0x7480, 0x74c0, 0x7500

callid_code = {}

callid_name = {}

def display_callid(addr, blk):

"""Decode CALL-ID table which spans four 64 byte transfers.

We assume here that the requests are in address order.

There are two codes/names per line as 0 and 1 but both names

and codes are 1-20

struct callid_7400_7440 {

struct {

unsigned char cid0[6];

unsigned char pad0[2];

unsigned char cid1[6];

unsigned char pad1[2];

} callids[4];

}

struct callid_7800 {

struct {

unsigned char cid0[6];

unsigned char pad0;

unsigned char cid1[6];

unsigned char pad1;

} callids[2];

struct {

char name0[6];

char pad0[2]

char name1[6];

char pad1[2]

} names[2]

}

struct callid_78c0_7500 {

struct {

char name0[6];

char pad0[2]

char name1[6];

char pad1[2]

} names[2]

}

"""

global callid_code

global callid_name

if (addr == 0x7400): # callid codes 1-8

callid_code = {}

callid_name = {}

for i in range(0,8):

cid = callid2str(blk[0:8])

callid_code[i] = cid

blk = blk[8:]

elif (addr == 0x7440): # callid codes 9-15

for i in range(8,16):

rec = blk[0:8]

cid = callid2str(blk[0:8])

callid_code[i] = cid

blk = blk[8:]

elif (addr == 0x7480): # callid codes 16-20, names 1-4

for i in range(16,20):

rec = blk[0:8]

cid = callid2str(blk[0:8])

callid_code[i] = cid

blk = blk[8:]

for i in range(0,4):

rec = blk[0:8]

name = struct.unpack(">6sxx", rec)

callid_name[i] = name[0]

blk = blk[8:]

elif (addr == 0x74c0): # names 5-13

for i in range(4,12):

rec = blk[0:8]

name = struct.unpack(">6sxx", rec)

callid_name[i] = name[0]

blk = blk[8:]

else: # 0x7500 names 14-20, print them.

for i in range(12,20):

rec = blk[0:8]

name = struct.unpack(">6sxx", rec)

callid_name[i] = name[0]

blk = blk[8:]

for i in range(0,20):

name = callid_name[i].replace('\0', ' ')

code = callid_code[i]

print("call-id %2d '%s': code '%s'" % (i+1, name, code))

# Command/reply type decodes and encodes

def display_config(addr, contents):

"""Decode config addr blobs into something useful/readable

"""

item_name = conf_item_name(addr)

print("config item %s (0x%x)" % (item_name, addr))

if (addr == 0x8b0):

display_radio_conf(addr, contents)

elif (addr == 0x8c0):

display_ptt_id(addr, contents)

elif (addr == 0x8d0):

display_scc_control(addr, contents)

elif (addr == 0x8f0): # no menu, Banner 16 bytes, null term text followed by space (0x20)

print("%s: '%s'" % (item_name, contents.replace('\0', ' '))) # make trailing nulls into spaces for pretty

elif (addr == 0x740 or addr == 0x750 or addr == 0x760):

display_fm_chans(addr, contents)

elif (addr == 0x940 or addr == 0x980):

display_scan_group(addr, contents)

elif (addr == 0x7400 or addr == 0x7440 or addr == 0x7480 or addr == 0x74c0 or addr == 0x7500):

display_callid(addr, contents)

else:

print("%s" % hexprint(contents))

def display_ident(blk):

"""The ident block identifies the radio and its capabilities. This block is always 78 bytes

The rev == '01' is the base radio and '02' seems to be the '-Plus' version

"""

fmt = '>7s2s'

rec = blk[0:9]

rest = blk[9:]

part1 = rest[0:15]

part2 = rest[15:28]

part21 = rest[28:33]

part3 = rest[33:48]

part4 = rest[48:61]

part41 = rest[61:]

(model, rev) = struct.unpack(fmt, rec)

print("model = '%s', revision = '%s'" % (model, rev))

print("part1 [%i]\n%s" % (len(part1), hexprint(part1)))

print("part2 [%i]\n%s" % (len(part2), hexprint(part2)))

print("part21 [%i]\n%s" % (len(part2), hexprint(part21)))

print("part3 [%i]\n%s" % (len(part3), hexprint(part3)))

print("part4 [%i]\n%s" % (len(part3), hexprint(part4)))

print("part41 [%i]\n%s" % (len(part3), hexprint(part41)))

# Channel memory read/write decode/encode

def display_chan_blk(address, blk):

"""The channel memory is composed of 96 byte blocks of the following

structure. Each block contains four memories. The blocks are sequencially

addressed to support the 1-999 memories.

struct chan_blk {

struct {

int32_t rx;

int32_t tx;

unsigned char[8];

} freq[4];

struct {

char name[8];

} name[8];

}

The radio deletes a channel by nulling the name and setting the receive

frequency to -1. tx freq is unchanged and bits partially change.

"""

if (len(blk) != 96):

print("display_chan_blk: block too small %d" % len(blk))

return

# First pick up the frequency and mode/modulation/??? records

freqs = {}

for i in range(0,4):

rec = blk[0:16]

(r, t, bitsh, bitsl) = struct.unpack('>llLL', rec)

freqs[i] = (r, t, bitsh, bitsl)

blk = blk[16:]

# Followed by the names

names = {}

for i in range(0,4):

rec = blk[0:8]

name = struct.unpack('8s', rec)

names[i] = name[0].replace('\0', ' ') # make trailing nulls into spaces for pretty

blk = blk[8:]

# Put it all together and print it

print("Address: 0x%x" % address)

for i in range(0,4):

r, t, bitsh, bitsl = freqs[i]

chan = ((address>>4) - 0xa0) + 1 +i # map memory address and record to radio channel number

print("rec %d, ch %3d [%s]: rx=%s, tx=%s, bits=%08x-%08x" % (i, chan, names[i], int2freq(r), int2freq(t),

bitsh, bitsl))

# Process a line to/from the radio

def _read_record(lineno, data):

"""Return valid, cmd, direction, size, payload """

global checksum_seed

errs = ""

### 7D 82 00 12 52 12 22 12 22 12 22 12 ED 12 22 12 22 12 22 12 ED 12 12

if (len(data) < 4):

print("Line %d: Error: Packet too short (%d)" % (lineno, len(data)))

return (False, 0, 0, 0, 0)

if (data[0] != 0x7D):

print("Line %d: Error: Packet start is not right. %02x != 0x7D" % (lineno, data[0]))

return (False, 0, 0, 0, 0)

# Decode the packet

(ckmatch, cmd, xfer_dir, _length, payload, decode_errs) = pkt_decode(data)

# Validations

# direction

if (xfer_dir == 0x00):

direction = "reply"

to_radio = False

elif (xfer_dir == 0xFF):

direction = "command"

to_radio = True

else:

print("Line %d: Unknown direction: %02x" % (lineno, xfer_dir))

return(False, 0, 0,0,0)

# Payload length

if (_length > 0 and ((len(payload)) != _length)): #

errs += "Packet length does not match len(payload) = %d, length = %d; " % (len(payload) , _length)

if (to_radio):

if (cmd in cmd_lerrs):

cmd_lerrs[cmd] += 1

else:

cmd_lerrs[cmd] = 1

else:

if (cmd in rep_lerrs):

rep_lerrs[cmd] += 1

else:

rep_lerrs[cmd] = 1

#Checksum match

if (not ckmatch):

errs += decode_errs

if (to_radio):

if (cmd in cmd_cerrs):

cmd_cerrs[cmd] += 1

else:

cmd_cerrs[cmd] = 1

else:

if (cmd in rep_cerrs):

rep_cerrs[cmd] += 1

else:

rep_cerrs[cmd] = 1

# Count commands and replys by type

if (to_radio):

if (cmd in cmd_cnt):

cmd_cnt[cmd] += 1

else:

cmd_cnt[cmd] = 1

else:

if (cmd in rep_cnt):

rep_cnt[cmd] += 1

else:

rep_cnt[cmd] = 1

# Decode packets

print("Line %d: %s %s (%d) %s" % (lineno, cmd_name[cmd], direction, _length, errs))

if (cmd == 0x80): # Hello, ident yourself

if (to_radio):

if(_length != 0):

print("malformed ident query")

else: # Hello, ident yourself

display_ident(payload)

elif (cmd == 0x81): # Hang up. We are done. No reply from radio

if (to_radio):

if(_length != 0):

print("malformed ident query")

else: # should get no hangup reply

print("Error: radio should reboot, not reply!")

else: # these commands move data

address = payload[0]<<8 | payload[1]

if (cmd == 0x82): # Read config

if to_radio:

count = payload[2]

print("config item %s (0x%x), fetch %d" % (conf_item_name(address), address, count))

else:

item_value = payload[2:]

display_config(address, item_value)

elif (cmd == 0x83): # write config

if to_radio:

item_value = payload[2:]

display_config(address, item_value)

else:

print("%s: OK" % conf_item_name(address))

elif (cmd == 0x84): # Read channel memory

if to_radio:

count = payload[2]

print("EEPROM addr 0x%x, fetch %d" % (address, count))

else:

eeprom_contents = payload[2:]

display_chan_blk(address, eeprom_contents)

elif (cmd == 0x85): # Write channel memory

if to_radio:

eeprom_contents = payload[2:]

display_chan_blk(address, eeprom_contents)

else:

print("%s: OK" % conf_item_name(address))

else:

print("Unknown %s command: %x\n[%i]\n%s" % (direction, cmd, len(payload), hexprint(payload)))

print("\n------------------ end of line ---------------------------")

# valid record

return (True, cmd, direction, _length, payload)

def l2b(data):

"""Line lo bytes string safely.

using chr() introduces unicode code point conversions.

We use (force) real byte (integer) types because duck typing of strings bite back."""

# convert the data in string format to unicode safe byte format

return bytearray.fromhex(data)

def dump(lineno, data):

(valid, cmd, direction, size, payload) = _read_record(lineno, l2b(data))

global cmd_cnt

global rep_cnt

global cmd_cerrs

global rep_cerrs

global cmd_lerrs

global rep_lerrs

# packet validation

if (valid == False):

# invalid packet, finish

print("Invalid packet at line %d" % lineno)

elif (cmd == 0x81): # roll up stats on hangup

print("At hangup at line %d. command stats\n # cmds cksm length" % lineno)

for cmd in list(cmd_cnt):

if (cmd in cmd_cerrs):

cerrs = cmd_cerrs[cmd]

else:

cerrs = 0

if (cmd in cmd_lerrs):

lerrs = cmd_lerrs[cmd]

else:

lerrs = 0

print("command %12s: %4d, %4d, %4d" % (cmd_name[cmd], cmd_cnt[cmd], cerrs, lerrs))

for cmd in list(rep_cnt):

if (cmd in rep_cerrs):

cerrs = rep_cerrs[cmd]

else:

cerrs = 0

if (cmd in rep_lerrs):

lerrs = rep_lerrs[cmd]

else:

lerrs = 0

print("reply %12s: %4d, %4d, %4d" % (cmd_name[cmd], rep_cnt[cmd], cerrs, lerrs))

cmd_cnt = {}

rep_cnt = {}

cmd_cerrs = {}

rep_cerrs = {}

cmd_lerrs = {}

rep_lerrs = {}

print("========================= end of session ====================================");

return cmd

# run it on the same folder as the data_down_clean.txt file

with open(sys.argv[1]) as f:

linenum = 1

for line in f:

line = line.replace('\n', '') # chomp

# drop empty lines and lines w/out the magic tag.

if (len(line) > 0 and line[0] in 'WR' and line[1] == ":"):

comm = dump(linenum, line[3:])

linenum = linenum + 1

## # dump just a few packets

## if (linenum > 3):

## break