syzkaller_drill.py

syzkaller_drill.py

from lxml import html
import argparse
import logging
import os
import re
import requests
import subprocess
import sys
import tempfile

CACHE_FILE = "/tmp/syzkaller.html"

if __name__ == "__main__":
    if not os.path.exists(CACHE_FILE):
        logging.info("Caching coverage HTML")
        req = requests.get("https://storage.googleapis.com/syzkaller/cover/ci-upstream-kasan-gce-root.html", stream=True)
        req.raise_for_status()
        with open(CACHE_FILE, 'wb') as f:
            for chunk in req.iter_content(1*1000*1000):
                f.write(chunk)

    parser = argparse.ArgumentParser()
    parser.add_argument("target_file")
    parser.add_argument("target_lineno", type=int)
    parser.add_argument("--syz-only", action="store_true", help="Return the syzkaller reproducer instead of the converted C")
    args = parser.parse_args()

    if not args.target_file.startswith('/'):
        args.target_file = '/' + args.target_file

    with open(CACHE_FILE, 'rb') as f:
        tree = html.parse(f)
        flink = tree.xpath(f"//a[@id='path{args.target_file}']")
        if not flink:
            logging.fatal("Unable to find file %s", args.target_file)
            sys.exit(1)
        fid = int(re.findall(r'onFileClick\(\s*(\d+)\s*\)', flink[0].attrib['onclick'])[0])
        logging.info("File %s is fid %s", args.target_file, fid)
        content = tree.xpath(f"//pre[@id='contents_{fid}']")[0]
        prog_refs, linenos, lines = content.getnext().iterdescendants('td')
        ref_elems = prog_refs.getchildren()
        for ref_str, lineno, content in zip(
                prog_refs.text_content().split('\n'),
                linenos.text_content().split('\n'),
                lines.text_content().split('\n')):
            lineno = int(lineno)

            ref_elem = None
            if ref_str.strip() != '':
                ref_elem = ref_elems.pop(0)

            if lineno == args.target_lineno:
                if ref_elem is None:
                    logging.fatal("Line %s has no program", args.target_lineno)
                    sys.exit(1)
                break
        else:
            logging.fatal("Unable to find line %s", args.target_lineno)
            sys.exit(1)

        prog_id = int(re.findall(r'onProgClick\((\d+), this\)', ref_elem.attrib['onclick'])[0])
        logging.info("Line %s has target program id %s", args.target_lineno, prog_id)

        program = str(tree.xpath(f"//pre[@id='prog_{prog_id}']")[0].text_content()).strip()
        if args.syz_only:
            print(program)
        else:
            with tempfile.NamedTemporaryFile('w+') as tmpf:
                tmpf.write(program)
                tmpf.flush()
                prog2c = os.path.join(os.environ.get("SYZ_BIN", ""), "syz-prog2c")
                print(subprocess.check_output([prog2c, "-prog", tmpf.name]).decode('utf-8'))