On the Wire

In this challenge, Even Booth is outside of City Hall asking for our help. He has this little gnome sensor that is reading various signals that are being broadcast by the gnomes around the neighborhood.

His sensor has a websocket that you can connect to to listen in on the live broadcast. These types of challenges help me not take for granted the tools I use everyday for reading and interpreting low level sensor data. There are 3 communication channels here to unpack, 1-wire, SPI, and i2c. I will check these one at a time. There are great hints provided on this. I try to avoid hints if I can help it but these proved to be very valuable.

So the first thing I did was write a python script to tap into the websock of this sensor and capture the broadcasts of all the signals. Some protocols have 2 channels, a data channel and a clock channel. It will record all the live data and save it offline in a file, so I don't have to continually listen to live data.

class HolidayHackClient(threading.Thread):
    def __init__(self, url="1w_dq", stop_str="", debug=False):
        super().__init__()
        self.urls = {
            "1w_dq": "wss://signals.holidayhackchallenge.com/wire/dq",
            "spi_mosi": "wss://signals.holidayhackchallenge.com/wire/mosi",
            "spi_sck": "wss://signals.holidayhackchallenge.com/wire/sck",
            "i2c_sda": "wss://signals.holidayhackchallenge.com/wire/sda",
            "i2c_scl": "wss://signals.holidayhackchallenge.com/wire/scl"
        }
        self.wsapp = websocket.WebSocketApp(
            self.urls[url], 
            on_open=self.on_open, 
            on_message=self.on_message)
        self.msgs = []
        self.stop_str = stop_str
        self.stopped = False
        self.debug = debug
        
    def on_open(self, wsapp):
        print("WS Connected")

    def on_message(self, wsapp, message):
        if self.debug:
            print(message)
        try:
            msg_obj = json.loads(message)
        except json.JSONDecodeError as e:
            print("Could not JSON Decode Message: ")
            print(e)
            return
        if msg_obj.get("type", "") == "welcome":
            return
        self.msgs.append(msg_obj)
        if self.stop_str in message:
            self.close() 

    def close(self):
        if not self.stopped:
            self.wsapp.close()
            self.stopped = True

    def send(self, msg):
        self.wsapp.send_text(msg)

    def run(self):
        print("Run")
        self.wsapp.run_forever()
    
    def get_msgs(self):
        return self.msgs

This class can connect to one of the websocket URL's and save messages as JSON until you press Ctrl+C or a stop string is detected. If you watch the bursts in real time, it always broadcasts the same data over and over. I just used that last message as a stop string so it knows when the burst is done.

I can then use a client function to make and record all the signals. I will post all this code in one file at the end.

def collect_msgs():
    all_msgs = {}
    urls = [
        ["1w_dq", "stop"],
        ["spi_mosi", "7990000"],
        ["spi_sck", "8000000"],
        ["i2c_sda", "2582000"],
        ["i2c_scl", "2580000"],
    ]
    for url in urls:
        try:
            client = HolidayHackClient(url[0], url[1], False)
            client.start()
            while not client.stopped:
                time.sleep(0.001)

            msgs = client.get_msgs()
            all_msgs[url[0]] = []
            # sort
            msgs = sorted(msgs, key=lambda d: d['t'])
            for msg in msgs:
                #print(msg)
                all_msgs[url[0]].append(msg)
        except KeyboardInterrupt as e:
            print("Closing connection")
            client.close()
        finally:
            client.join()
            print("Done")
    with open("all_msgs.json", "w") as f:
        json.dump(all_msgs, f, indent=2)

This function collects websocket messages, sorts them based on timestamp (important), then saves the data in JSON format in a file "all_messages.json", where all the channels and messages can be accessed in one file. If you want to check it out here it is:

all_msgs.json

This JSON file is structured in an object where the top level key names denote the channel, and their values are arrays of the data objects.

Now that I have all these messages save, I don't have to connect to the websocket anymore.

The first protocol to tackle is 1-wire. This protocol is widely used by sensors and small devices to commicate data. It is described well in the 1-Wire Wiki. This protocol only has a data line, and it can only be high and low. Check out what the hint provides us:

For this protocol, the distances between the voltage changes are the important bit. I started measuring the pulse starting when I detected a falling edge, aka high -> low. Measuring all the widths I noticed they were either 6μs or 60μs. This makes sense because in the Wikipedia article, they reference this spacing: "Therefore, the "0" pulses have to be 60 μs long, and the "1" pulses can't be longer than 15 μs." I also noticed the first pulse width is 480μs which the wikipedia calls a "reset" pulse. This denotes the beginning of the transaction. This is why I don't start reading from the very first message. Check out my decode script below.

def decode_1w(msgs):
    msgs = msgs[4:]
    start = False
    bs = ""
    # Interpret the pulses into a string of bits.
    pulse_start = msgs[0]["t"]
    for x in range(1,len(msgs)):
        if msgs[x]["v"] == msgs[x-1]["v"]:
            continue
        if msgs[x]["v"] == 1 and msgs[x-1]["v"] == 0:
            # Rising Edge
            if msgs[x]["t"]-pulse_start < 15:
                # High Pulse: 1
                bs += "1"
            else:
                # Low Pulse: 0
                bs += "0"
        if msgs[x]["v"] == 0 and msgs[x-1]["v"] == 1:
            # Falling Edge 
            pulse_start = msgs[x]["t"]
    
    s = ""
    # Read the bits right to left - LSB
    for x in range(0, len(bs), 8):
        bits = bs[x:x+8]
        bit_int = int(bits[::-1], 2)
        s+=chr(bit_int)
    print(s)

This code reads the pulse widths and interprets them into a string of bits. Then reads the string 8 bits at a time, interpreting them as ASCII. Then it prints it out. Remember to consider that the bits could be read left to right (Most Significant Bit) or right to left (Least Significant Bit).

read and decrypt the SPI bus data using the XOR key: icy

I got a message! This tells us that the data on the SPI bus needs to be decoded using XOR code key "icy". Lets go.

SPI or Serial Peripheral Interface is another data communication protocol. Checkout the SPI Wiki for all the information. This one uses 2 channels, a data channel (MOSI - Master Out/Slave In) and a clock channel (SCLK). The hint really help interpreting this clock and data line:

Since I already have a sorted list of messages, I just need to read the clock line and watch for rising or falling edges. To prepare for this, I wrote a function to take the data line and look up weather the line is high or low at that timestamp.

def value_at(data_msgs, timestamp):
    for x in range(1, len(data_msgs)):
        cur = data_msgs[x]['t']
        prev = data_msgs[x-1]['t']
        if timestamp >= prev and timestamp < cur:
            #print(f"{prev} <= {timestamp} < {cur}")
            return data_msgs[x-1]["v"]
    #print(f"{timestamp} >= {data_msgs[-1]['t']}")
    return data_msgs[-1]['v']

It just returns the value (1 or 0) at that particular timestamp. One other bit of preparation for this one is I need a function to decode bytes with an XOR key.

def xor_decrypt(encrypted_data: bytes, key: bytes) -> bytes:
    decrypted_bytes = bytes(a ^ b for a, b in zip(encrypted_data, itertools.cycle(key)))
    return decrypted_bytes

It just reads the input byte by byte doing an XOR with the current key byte, which it cycles through. Now I'm ready for the actual data. I can just go along the clock line and detect edges/timestamps to build a bit string. At the end I can use the "xor_decrypt" function to get the plaintext of the message.

def decode_spi(data_msgs, clock_msgs):
    data_msgs = data_msgs[1:]
    bs = ""
    for msg in clock_msgs:
        if msg.get("marker", "") != "sample":
            continue
        v = value_at(data_msgs, msg["t"])
        bs += str(v)

    int_list = []
    # convert bit string into list of ints
    for x in range(0, len(bs), 8):
        bits = bs[x:x+8]
        bit_int = int(bits, 2)
        int_list.append(bit_int)

    key = b"icy"
    s = xor_decrypt(int_list, key)
    print(s.decode('ascii'))

I ran this function and got some clear text!

read and decrypt the I2C bus data using the XOR key: bananza. the temperature sensor address is 0x3C

Now I know the next XOR key and I know which address to look for in the i2C data!

For I2C, we need to think about the data in transactions. A transaction is a series of bytes, where the first byte is a read/write bit followed by a 7-bit address. The following bytes are the data.

So I2C was a little strange because the data line contains so much metadata that the clock line isn't necessary. I know this isn't how I2C isn't supposed to be read, but just check out the data line:

{'line': 'sda', 't': 4000, 'v': 1, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 0, 'type': 'address'}
{'line': 'sda', 't': 14000, 'v': 0, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 1, 'type': 'address'}
{'line': 'sda', 't': 24000, 'v': 0, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 2, 'type': 'address'}
{'line': 'sda', 't': 34000, 'v': 1, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 3, 'type': 'address'}
{'line': 'sda', 't': 44000, 'v': 0, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 4, 'type': 'address'}
{'line': 'sda', 't': 54000, 'v': 0, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 5, 'type': 'address'}
{'line': 'sda', 't': 64000, 'v': 0, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 6, 'type': 'address'}
{'line': 'sda', 't': 74000, 'v': 0, 'marker': 'address-bit', 'byteIndex': 0, 'bitIndex': 7, 'type': 'address'}
{'line': 'sda', 't': 84000, 'v': 0, 'marker': 'ack-bit', 'byteIndex': 0, 'type': 'ack'}
{'line': 'sda', 't': 94000, 'v': 1, 'marker': 'ack-release', 'byteIndex': 0, 'type': 'ack'}
{'line': 'sda', 't': 94000, 'v': 0, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 0, 'type': 'data'}
{'line': 'sda', 't': 104000, 'v': 1, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 1, 'type': 'data'}
{'line': 'sda', 't': 114000, 'v': 0, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 2, 'type': 'data'}
{'line': 'sda', 't': 124000, 'v': 1, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 3, 'type': 'data'}
{'line': 'sda', 't': 134000, 'v': 0, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 4, 'type': 'data'}
{'line': 'sda', 't': 144000, 'v': 1, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 5, 'type': 'data'}
{'line': 'sda', 't': 154000, 'v': 1, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 6, 'type': 'data'}
{'line': 'sda', 't': 164000, 'v': 0, 'marker': 'data-bit', 'byteIndex': 1, 'bitIndex': 7, 'type': 'data'}
{'line': 'sda', 't': 174000, 'v': 0, 'marker': 'ack-bit', 'byteIndex': 1, 'type': 'ack'}
{'line': 'sda', 't': 184000, 'v': 1, 'marker': 'ack-release', 'byteIndex': 1, 'type': 'ack'}

The "type" field tells us if its the address bit or data bit, the byte index tells us the index of the byte in the transaction, and the bit index tells us the index of the bit for that particular byte in the transactions. This is all the data you need to construct these transactions. No clock line necessary! Remember these messages are pre-sorted by timestamp.

def decode_i2c(data_msgs, clock_msgs):
    transactions = []
    transaction = []
    
    # Read the data, putting the bits in the correct indexes
    # based on the byteIndex,bitIndex fields
    for msg in data_msgs:
        if msg["marker"] == "start":
            transaction = []
        if msg["marker"] in ["address-bit", "data-bit"]:
            if len(transaction) <= msg["byteIndex"]:
                transaction.append(list("00000000"))
            transaction[msg["byteIndex"]][msg["bitIndex"]] = msg["v"]
        if msg["marker"] == "stop":
            transactions.append(transaction)

    # Construct transaction objects (address,data)
    transaction_objs = []
    for t in transactions:
        # right-most 7 bits are the address
        address = hex(int(''.join(map(str,t[0]))[:-1], 2))
        obj = {
            "address": address,
            "data": []
        }
        for x in range(1, len(t)):
            obj["data"].append(int(''.join(map(str,t[x])), 2))
        transaction_objs.append(obj)
    
    # We only care about address 0x3C
    for obj in transaction_objs:
        if obj["address"] != "0x3c":
            continue
        s = xor_decrypt(obj["data"], b"bananza")
        print(s.decode("ascii"))

So I constructed all the transactions, looked for the one at address 0x3C. I can then perform that same XOR decrypt using the key "bonanza". I printed it out and got a clear text temperature reading!

32.84

Enter this in your badge to complete the On the Wire challenge!

As promised, the full script can be downloaded here:
on_the_wire.py