Fuzzer for rtl_433

fuzz.py

#!/usr/bin/env python3

"""
Fuzz.py, fuzzing the rtl_433 decoders for possible errors.
Copyright (C) 2019  Rick van Schijndel

This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; either version 2
of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
"""

"""
This application generates random 'signals' and feeds them to rtl_433 using the -y test option.
This has proven to be useful to find decoders that match too often on data that's invalid.
Some decoders can't be fixed as they don't have a checksum,
but others might miss some checksum checks or have other easily-fixable problems.
"""

import collections
import json
import subprocess
import random
from binascii import hexlify

def generate_signal():
    """
    Generates a random 'test signal' using the format {nbits}data (see rtl_433 -h for more info).
    """
    amount = random.randint(1, 254)
    data = bytes(random.sample(range(0xff), amount))
    data_hex = bytes.hex(data)
    return '{{{}}}{}'.format(amount, str(data_hex))


def run(input):
    """
    Run rtl_433 with the given input as test signal.
    
    stdout and stderr are recorded using a pipe and can be retrieved from the return value.
    """
    return subprocess.run(['rtl_433', '-y', input, '-M', 'newmodel', '-F', 'json'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)


def main():
    found = 0
    # How much signal should be generated before giving the results
    total = 30000 

    devices = collections.defaultdict(int)

    for i in range(total):
        generated = generate_signal()
        output = run(generated)
    
        # If rtl_433 gives any output on stdout, this means a decoder found data.
        if output.stdout:
            # Print progress if any new match if found.
            print('{}/{}'.format(i, total), end=' ')
            print(generated)
            decoded = bytes.decode(output.stdout).strip().split('\n')
            for line in decoded:
                data = json.loads(line)
                # Count this device as a match.
                devices[data['model']] += 1
                # Print the data that has been found.
                # This can be useful for finding decoders that give bogus data,
                # such as humidity above 100% or so.
                #print(json.dumps(data, indent=4, sort_keys=True))
            found += 1
   
    print('count of devices triggered using random data (high count may mean the device triggers too often)')
    for device in sorted(devices, key=devices.get, reverse=True):
        count = devices[device]
        print(count, device, sep='\t')

    print('Found {} signals ({:.2f}% of all tests found a signal)'.format(found, (found / total * 100)))

if __name__ == '__main__':
    main()