urchin

#!/usr/bin/env python3
"""
Verify script for Sea Urchin verification jobs
return code
0 - all ok
1 - expected string not found in stderr/stdout
2 - error reported to stderr
"""
import argparse
import json
import os
import os.path
import re
import sys

quiet = False

SEAHORN_ROOT = os.environ['SEAHORN_ROOT']
ASSERT_ERROR_PREFIX = r'^Error: assertion failed'
# the plan is to have two sets, vac error and info and put filepath:linenumbers) into both
VACUITY_CHECK_RE = r'^(?P<stream>Info|Error).*(?P<what>vacuity).*(?P<result>passed|failed).*sat\) (?P<debuginfo>.*)$'


def check_vacuity(line, passed_set, failed_set):
    m = re.match(VACUITY_CHECK_RE, line)
    if not m:
        return
    debugInfo = m.group('debuginfo')
    if (m.group('stream') == 'Info'
            and m.group('what') == 'vacuity'
            and m.group('result') == 'passed'):
        passed_set.add(debugInfo.strip())
    elif (m.group('stream') == 'Error'
          and m.group('what') == 'vacuity'
          and m.group('result') == 'failed'):
        failed_set.add(debugInfo.strip())


def main(argv):
    import sea

    def check_vacuity_inner(line, passed_set, failed_set):
        return check_vacuity(line, passed_set, failed_set)

    def _add_S_arg(ap):
        ap.add_argument('-S', dest='llvm_asm', default=False, action='store_true',
                        help='Write output as LLVM assembly')
        return ap

    def _bc_or_ll_file(name):
        ext = os.path.splitext(name)[1]
        return ext == '.bc' or ext == '.ll'

    class Cargo(sea.LimitedCmd):
        """
        Produce a single bitcode for a Cargo(Rust) project.
        If the input file is already bitcode then complain.
        """

        def __init__(self, quiet=False):
            super(Cargo, self).__init__('cargo', 'Compile Rust', allow_extra=True)
            self.cargoCmd = None

        def mk_arg_parser(self, ap):
            ap = super(Cargo, self).mk_arg_parser(ap)
            ap.add_argument('--verbose', '-v', dest='verbose', type=int, default=0,
                            help='Verbosity level', metavar='N')
            ap.add_argument('--test', dest='test_name',
                            metavar='TEST_NAME', help='Name of Test', default=None)
            ap.add_argument('--local', action='store_true', default=False,
                              help='Use local cargo-verify')
            sea.add_tmp_dir_args(ap)
            sea.add_in_out_args(ap)
            _add_S_arg(ap)
            return ap

        def _construct_out_filename(self, in_file, ext, work_dir):
            """
            Construct an out filename

            :param in_file: the input file path
            :param ext: the extension of the output file
            :param work_dir: the workdir is ignored
            :return: the output file path. It is in the 'target' dir of Cargo project
            """
            dir, base = os.path.split(in_file)
            if os.path.isdir(in_file):
                dir = in_file
                base = ''
            if base == '':
                base = os.path.basename(dir)
            out_base = os.path.splitext(base)[0] + ext
            out_file = os.path.join(dir, 'target', out_base)
            return os.path.abspath(out_file)

        def name_out_file(self, in_files, args=None, work_dir=None):
            in_file = in_files[0]
            ext = '.rvt.json'
            return self._construct_out_filename(in_file, ext, work_dir)

        def run(self, args, extra):
            out_file = args.out_file
            if len(args.in_files) != 1:
                raise IOError('Cargo only accepts one input file')
            # Error on .bc and .ll files
            if _bc_or_ll_file(args.in_files[0]):
                raise IOError('Cargo does not accept ll/bc file')

            argv = list()
            if args.local:
                cmd_name = sea.which(['cargo-verify'])
            else:
                script_dir = os.path.abspath(sys.argv[0])
                script_dir = os.path.dirname(script_dir)
                cmd_name = os.path.join(script_dir, 'cargo-verify')
                # cmd to pass to docker container
                argv.extend(['cargo-verify'])

            if cmd_name is None: raise IOError('cargo-verify not found')
            self.cargoCmd = sea.ExtCmd(cmd_name, '', quiet)

            if out_file is not None:
                argv.extend(['-o', out_file])
            input_file =  os.path.abspath(args.in_files[0])
            if os.path.isdir(input_file):
                _input_file = os.path.join(input_file, 'Cargo.toml')
                if os.path.isfile(_input_file):
                    input_file = _input_file
                else:
                    raise IOError('Cargo only accepts dir name which contain Cargo.toml')
            else:
                raise IOError('Cargo only accepts dir name which contain Cargo.toml')
            argv.extend(['--manifest-path', input_file])
            argv.extend(['--backend', 'seahorn'])
            argv.extend(['--dry-run'])
            argv.extend(['--clean'])
            if args.test_name:
                argv.extend(['--test', args.test_name])
            if args.verbose > 0:
                argv.extend(['-vv'])
            return self.cargoCmd.run(args, argv)

        @property
        def stdout(self):
            return self.cargoCmd.stdout

    class UrchinVerify(sea.LimitedCmd):
        def __init__(self, seaSeqCmd, quiet=False):
            super(UrchinVerify, self).__init__('urchin', 'Get bitcode and entry point for Cargo project',
                                               allow_extra=True)
            self.seaSeqCmd = seaSeqCmd

        def mk_arg_parser(self, ap):
            class EnhancedArgParser(argparse.ArgumentParser):
                """
                This ArgParser overrides print_help() behaviour to print help of contained cmd.
                """
                def __init__(self, prog, description='', add_help=False, subc=None):
                    super().__init__(prog=prog, description=description, add_help=add_help)
                    self.subc = subc

                def print_help(self):
                    super().print_help()
                    if self.subc is not None:
                        apv = argparse.ArgumentParser(self.subc.name,
                                                      description=self.subc.help,
                                                      add_help=False)
                        apv = self.subc.mk_arg_parser(apv)
                        apv.print_help()

            ap = EnhancedArgParser(self.name, description=self.help, add_help=False, subc=self.seaSeqCmd)
            ap = super(UrchinVerify, self).mk_arg_parser(ap)
            ap.add_argument('--test', dest='test_name',
                            metavar='TEST_NAME', help='Name of Test', default=None)
            # ap.add_argument('extra', nargs=argparse.REMAINDER)
            sea.add_tmp_dir_args(ap)
            sea.add_in_out_args(ap)
            return ap

        def name_out_file(self, in_files, args=None, work_dir=None):
            return self.seaSeqCmd.name_out_file(in_files, args, work_dir)

        def run(self, args, extra):
            entry_point = None
            seapp_inp_file = None
            with open(args.in_files[0]) as f:
                data = json.load(f)
                seapp_inp_file = data['bc_path']
                unmangled = args.test_name if args.test_name is not None else 'main'
                for ep in data['entry_points']:
                    if ep['unmangled'] == unmangled:
                        entry_point = ep['mangled']
                        found_main = True
                        break
            if entry_point is None: raise IOError('UrchinPp: Entry point not found')
            if seapp_inp_file is None: raise IOError('UrchinPp: bc file entry not found')
            argv = list()
            argv.extend(['--entry=%s' % entry_point])
            argv.extend(extra)
            argv.extend([seapp_inp_file])
            return self.seaSeqCmd.main(argv)

    class SeaVerifyCmd(sea.LimitedCmd):
        def __init__(self):
            super().__init__('SeaVerify', 'verify using sea', allow_extra=True)

        def mk_arg_parser(self, argp):
            import argparse
            argp = super().mk_arg_parser(argp)
            argp.add_argument('--verbose', '-v', dest='verbose', type=int, default=0,
                              help='Verbosity level', metavar='N')
            argp.add_argument('--silent', action='store_true', default=False,
                              help='Do not produce any output')
            argp.add_argument('--expect', type=str, default=None,
                              help='Expected string in the output')
            argp.add_argument('--command', type=str, default='bpf',
                              help='sea command')
            argp.add_argument('--cex', action='store_true', default=False,
                              help='Counterexample mode')
            argp.add_argument('--vac', action='store_true', default=False,
                              help='Vacuity mode')
            argp.add_argument('input_file', nargs=1)
            argp.add_argument('--dry-run', dest='dry_run',
                              action='store_true', default=False,
                              help='Pass --dry-run to yama')
            argp.add_argument('extra', nargs=argparse.REMAINDER)
            return argp

        def run(self, args=None, _extra=[]):
            extra = _extra + args.extra

            script_dir = os.path.abspath(sys.argv[0])
            script_dir = os.path.dirname(script_dir)

            input_file = os.path.abspath(args.input_file[0])

            # try to guess input file from directory name
            if os.path.isdir(input_file):
                fname = os.path.basename(input_file)
                _input_file = os.path.join(input_file, 'llvm-ir', fname + '.ir',
                                           fname + '.ir.bc')
                if os.path.isfile(_input_file):
                    input_file = _input_file

            file_dir = input_file
            file_dir = os.path.dirname(file_dir)

            cmd = [os.path.join(SEAHORN_ROOT, 'bin', 'sea'),
                   'yama', '--yforce']

            # base config
            base_config = os.path.join(script_dir, 'seahorn', 'sea.yaml')
            if args.cex:
                base_config = os.path.join(script_dir, 'seahorn',
                                           'sea_cex.yaml')
            cmd.extend(['-y', base_config])
            # vacuity config
            if args.vac:
                vac_config = os.path.join(script_dir, 'seahorn',
                                          'sea_vac.yaml')
                cmd.extend(['-y', vac_config])

            # job specific config
            job_config = os.path.abspath(os.path.join(file_dir, '..', '..',
                                                      'sea.yaml'))
            cmd.extend(['-y', job_config])

            if args.dry_run:
                cmd.append('--dry-run')

            cmd.append(args.command)
            cmd.extend(extra)
            cmd.append(input_file)

            if args.verbose > 0:
                print(' '.join(cmd))
                cmd.append('-v%d' % args.verbose)

            if args.expect is None:
                os.execv(cmd[0], cmd)

            import subprocess
            process = subprocess.Popen(cmd, shell=False,
                                       encoding='utf-8',
                                       errors='ignore',
                                       stdout=subprocess.PIPE,
                                       stderr=subprocess.STDOUT)
            found_expected = False
            found_error = False
            vacuity_passed = set()
            vacuity_failed = set()
            for line in iter(process.stdout.readline, ''):
                if not args.silent:
                    print(line, end='')

                # checks after this line are mutually exclusive
                if not found_expected and args.expect is not None and line.strip() == args.expect:
                    found_expected = True
                elif re.match(ASSERT_ERROR_PREFIX, line):
                    found_error = True
                else:
                    check_vacuity_inner(line, vacuity_passed, vacuity_failed)

            process.stdout.close()
            rcode = process.wait()

            if args.vac and found_error:
                return 2
            elif args.vac and (vacuity_failed - vacuity_passed):
                return 2
            elif rcode == 0 and args.expect is not None:
                return 0 if found_expected else 1
            else:
                return rcode

    rpf = sea.SeqCmd('rpf', 'alias for Cargo|UrchinVerify', [Cargo(),
                                                             UrchinVerify(SeaVerifyCmd())])
    cmd = sea.AgregateCmd('urchin', 'Urchin Verification Framework', [rpf])

    # read extra flags from environment variable
    if 'VERIFY_FLAGS' in os.environ:
        env_flags = os.environ['VERIFY_FLAGS']
        env_flags = env_flags.split()
        argv = env_flags + argv

    return cmd.main(argv)


if __name__ == '__main__':
    root = os.path.abspath(SEAHORN_ROOT)
    bin_dir = os.path.join(root, 'bin')
    if os.path.isdir(bin_dir):
        os.environ['PATH'] = bin_dir + os.pathsep + os.environ['PATH']

    seapy_dir = os.path.join(root, 'lib', 'seapy')
    if os.path.isdir(seapy_dir):
        sys.path.insert(0, seapy_dir)

    sys.exit(main(sys.argv[1:]))