putty/test/testcrypt.py

import sys
import os
import numbers
import subprocess
import re
import string
import struct
from binascii import hexlify

assert sys.version_info[:2] >= (3,0), "This is Python 3 code"

# Expect to be run from the 'test' subdirectory, one level down from
# the main source
putty_srcdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

def coerce_to_bytes(arg):
    return arg.encode("UTF-8") if isinstance(arg, str) else arg

class ChildProcessFailure(Exception):
    pass

class ChildProcess(object):
    def __init__(self):
        self.sp = None
        self.debug = None
        self.exitstatus = None
        self.exception = None

        dbg = os.environ.get("PUTTY_TESTCRYPT_DEBUG")
        if dbg is not None:
            if dbg == "stderr":
                self.debug = sys.stderr
            else:
                sys.stderr.write("Unknown value '{}' for PUTTY_TESTCRYPT_DEBUG"
                                 " (try 'stderr'\n")
    def start(self):
        assert self.sp is None
        override_command = os.environ.get("PUTTY_TESTCRYPT")
        if override_command is None:
            cmd = [os.path.join(putty_srcdir, "testcrypt")]
            shell = False
        else:
            cmd = override_command
            shell = True
        self.sp = subprocess.Popen(
            cmd, shell=shell, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    def write_line(self, line):
        if self.exception is not None:
            # Re-raise our fatal-error exception, if it previously
            # occurred in a context where it couldn't be propagated (a
            # __del__ method).
            raise self.exception
        if self.debug is not None:
            self.debug.write("send: {}\n".format(line))
        self.sp.stdin.write(line + b"\n")
        self.sp.stdin.flush()
    def read_line(self):
        line = self.sp.stdout.readline()
        if len(line) == 0:
            self.exception = ChildProcessFailure("received EOF from testcrypt")
            raise self.exception
        line = line.rstrip(b"\r\n")
        if self.debug is not None:
            self.debug.write("recv: {}\n".format(line))
        return line
    def already_terminated(self):
        return self.sp is None and self.exitstatus is not None
    def funcall(self, cmd, args):
        if self.sp is None:
            assert self.exitstatus is None
            self.start()
        self.write_line(coerce_to_bytes(cmd) + b" " + b" ".join(
            coerce_to_bytes(arg) for arg in args))
        argcount = int(self.read_line())
        return [self.read_line() for arg in range(argcount)]
    def wait_for_exit(self):
        if self.sp is not None:
            self.sp.stdin.close()
            self.exitstatus = self.sp.wait()
            self.sp = None
    def check_return_status(self):
        self.wait_for_exit()
        if self.exitstatus is not None and self.exitstatus != 0:
            raise ChildProcessFailure("testcrypt returned exit status {}"
                                      .format(self.exitstatus))

childprocess = ChildProcess()

method_prefixes = {
    'val_wpoint': ['ecc_weierstrass_'],
    'val_mpoint': ['ecc_montgomery_'],
    'val_epoint': ['ecc_edwards_'],
    'val_hash': ['ssh_hash_'],
    'val_mac': ['ssh2_mac_'],
    'val_key': ['ssh_key_'],
    'val_cipher': ['ssh_cipher_'],
    'val_dh': ['dh_'],
    'val_ecdh': ['ssh_ecdhkex_'],
    'val_rsakex': ['ssh_rsakex_'],
    'val_prng': ['prng_'],
    'val_pcs': ['pcs_'],
    'val_pockle': ['pockle_'],
    'val_ntruencodeschedule': ['ntru_encode_schedule_', 'ntru_'],
}
method_lists = {t: [] for t in method_prefixes}

checked_enum_values = {}

class Value(object):
    def __init__(self, typename, ident):
        self._typename = typename
        self._ident = ident
        for methodname, function in method_lists.get(self._typename, []):
            setattr(self, methodname,
                    (lambda f: lambda *args: f(self, *args))(function))
    def _consumed(self):
        self._ident = None
    def __repr__(self):
        return "Value({!r}, {!r})".format(self._typename, self._ident)
    def __del__(self):
        if self._ident is not None and not childprocess.already_terminated():
            try:
                childprocess.funcall("free", [self._ident])
            except ChildProcessFailure:
                # If we see this exception now, we can't do anything
                # about it, because exceptions don't propagate out of
                # __del__ methods. Squelch it to prevent the annoying
                # runtime warning from Python, and the
                # 'self.exception' mechanism in the ChildProcess class
                # will raise it again at the next opportunity.
                #
                # (This covers both the case where testcrypt crashes
                # _during_ one of these free operations, and the
                # silencing of cascade failures when we try to send a
                # "free" command to testcrypt after it had already
                # crashed for some other reason.)
                pass
    def __long__(self):
        if self._typename != "val_mpint":
            raise TypeError("testcrypt values of types other than mpint"
                            " cannot be converted to integer")
        hexval = childprocess.funcall("mp_dump", [self._ident])[0]
        return 0 if len(hexval) == 0 else int(hexval, 16)
    def __int__(self):
        return int(self.__long__())

def marshal_string(val):
    val = coerce_to_bytes(val)
    assert isinstance(val, bytes), "Bad type for val_string input"
    return "".join(
        chr(b) if (0x20 <= b < 0x7F and b != 0x25)
        else "%{:02x}".format(b)
        for b in val)

def make_argword(arg, argtype, fnname, argindex, argname, to_preserve):
    typename, consumed = argtype
    if typename.startswith("opt_"):
        if arg is None:
            return "NULL"
        typename = typename[4:]
    if typename == "val_string":
        retwords = childprocess.funcall("newstring", [marshal_string(arg)])
        arg = make_retvals([typename], retwords, unpack_strings=False)[0]
        to_preserve.append(arg)
    if typename == "val_mpint" and isinstance(arg, numbers.Integral):
        retwords = childprocess.funcall("mp_literal", ["0x{:x}".format(arg)])
        arg = make_retvals([typename], retwords)[0]
        to_preserve.append(arg)
    if isinstance(arg, Value):
        if arg._typename != typename:
            raise TypeError(
                "{}() argument #{:d} ({}) should be {} ({} given)".format(
                fnname, argindex, argname, typename, arg._typename))
        ident = arg._ident
        if consumed:
            arg._consumed()
        return ident
    if typename == "uint" and isinstance(arg, numbers.Integral):
        return "0x{:x}".format(arg)
    if typename == "boolean":
        return "true" if arg else "false"
    if typename in {
            "hashalg", "macalg", "keyalg", "cipheralg",
            "dh_group", "ecdh_alg", "rsaorder", "primegenpolicy",
            "argon2flavour", "fptype", "httpdigesthash"}:
        arg = coerce_to_bytes(arg)
        if isinstance(arg, bytes) and b" " not in arg:
            dictkey = (typename, arg)
            if dictkey not in checked_enum_values:
                retwords = childprocess.funcall("checkenum", [typename, arg])
                assert len(retwords) == 1
                checked_enum_values[dictkey] = (retwords[0] == b"ok")
            if checked_enum_values[dictkey]:
                return arg
    if typename == "mpint_list":
        sublist = [make_argword(len(arg), ("uint", False),
                                fnname, argindex, argname, to_preserve)]
        for val in arg:
            sublist.append(make_argword(val, ("val_mpint", False),
                                        fnname, argindex, argname, to_preserve))
        return b" ".join(coerce_to_bytes(sub) for sub in sublist)
    if typename == "int16_list":
        sublist = [make_argword(len(arg), ("uint", False),
                                fnname, argindex, argname, to_preserve)]
        for val in arg:
            sublist.append(make_argword(val & 0xFFFF, ("uint", False),
                                        fnname, argindex, argname, to_preserve))
        return b" ".join(coerce_to_bytes(sub) for sub in sublist)
    raise TypeError(
        "Can't convert {}() argument #{:d} ({}) to {} (value was {!r})".format(
            fnname, argindex, argname, typename, arg))

def unpack_string(identifier):
    retwords = childprocess.funcall("getstring", [identifier])
    childprocess.funcall("free", [identifier])
    return re.sub(b"%[0-9A-F][0-9A-F]",
                  lambda m: bytes([int(m.group(0)[1:], 16)]),
                  retwords[0])

def unpack_mp(identifier):
    retwords = childprocess.funcall("mp_dump", [identifier])
    childprocess.funcall("free", [identifier])
    return int(retwords[0], 16)

def make_retval(rettype, word, unpack_strings):
    if rettype.startswith("opt_"):
        if word == b"NULL":
            return None
        rettype = rettype[4:]
    if rettype == "val_string" and unpack_strings:
        return unpack_string(word)
    if rettype == "val_keycomponents":
        kc = {}
        retwords = childprocess.funcall("key_components_count", [word])
        for i in range(int(retwords[0], 0)):
            args = [word, "{:d}".format(i)]
            retwords = childprocess.funcall("key_components_nth_name", args)
            kc_key = unpack_string(retwords[0])
            retwords = childprocess.funcall("key_components_nth_str", args)
            if retwords[0] != b"NULL":
                kc_value = unpack_string(retwords[0]).decode("ASCII")
            else:
                retwords = childprocess.funcall("key_components_nth_mp", args)
                kc_value = unpack_mp(retwords[0])
            kc[kc_key.decode("ASCII")] = kc_value
        childprocess.funcall("free", [word])
        return kc
    if rettype.startswith("val_"):
        return Value(rettype, word)
    elif rettype == "int" or rettype == "uint":
        return int(word, 0)
    elif rettype == "boolean":
        assert word == b"true" or word == b"false"
        return word == b"true"
    elif rettype in {"pocklestatus", "mr_result"}:
        return word.decode("ASCII")
    elif rettype == "int16_list":
        return list(map(int, word.split(b',')))
    raise TypeError("Can't deal with return value {!r} of type {!r}"
                    .format(word, rettype))

def make_retvals(rettypes, retwords, unpack_strings=True):
    assert len(rettypes) == len(retwords) # FIXME: better exception
    return [make_retval(rettype, word, unpack_strings)
            for rettype, word in zip(rettypes, retwords)]

class Function(object):
    def __init__(self, fnname, rettypes, retnames, argtypes, argnames):
        self.fnname = fnname
        self.rettypes = rettypes
        self.retnames = retnames
        self.argtypes = argtypes
        self.argnames = argnames
    def __repr__(self):
        return "<Function {}({}) -> ({})>".format(
            self.fnname,
            ", ".join(("consumed " if c else "")+t+" "+n
                       for (t,c),n in zip(self.argtypes, self.argnames)),
            ", ".join((t+" "+n if n is not None else t)
                      for t,n in zip(self.rettypes, self.retnames)),
    )
    def __call__(self, *args):
        if len(args) != len(self.argtypes):
            raise TypeError(
                "{}() takes exactly {} arguments ({} given)".format(
                    self.fnname, len(self.argtypes), len(args)))
        to_preserve = []
        retwords = childprocess.funcall(
            self.fnname, [make_argword(args[i], self.argtypes[i],
                                       self.fnname, i, self.argnames[i],
                                       to_preserve)
                          for i in range(len(args))])
        retvals = make_retvals(self.rettypes, retwords)
        if len(retvals) == 0:
            return None
        if len(retvals) == 1:
            return retvals[0]
        return tuple(retvals)

def _lex_testcrypt_header(header):
    pat = re.compile(
        # Skip any combination of whitespace and comments
        '(?:{})*'.format('|'.join((
            '[ \t\n]',             # whitespace
            '/\\*(?:.|\n)*?\\*/',  # C90-style /* ... */ comment, ended eagerly
            '//[^\n]*\n',          # C99-style comment to end-of-line
        ))) +
        # And then match a token
        '({})'.format('|'.join((
            # Punctuation
            r'\(',
            r'\)',
            ',',
            # Identifier
            '[A-Za-z_][A-Za-z0-9_]*',
            # End of string
            '$',
        )))
    )

    pos = 0
    end = len(header)
    while pos < end:
        m = pat.match(header, pos)
        assert m is not None, (
            "Failed to lex testcrypt-func.h at byte position {:d}".format(pos))

        pos = m.end()
        tok = m.group(1)
        if len(tok) == 0:
            assert pos == end, (
                "Empty token should only be returned at end of string")
        yield tok, m.start(1)

def _parse_testcrypt_header(tokens):
    def is_id(tok):
        return tok[0] in string.ascii_letters+"_"
    def expect(what, why, eof_ok=False):
        tok, pos = next(tokens)
        if tok == '' and eof_ok:
            return None
        if hasattr(what, '__call__'):
            description = lambda: ""
            ok = what(tok)
        elif isinstance(what, set):
            description = lambda: " or ".join("'"+x+"' " for x in sorted(what))
            ok = tok in what
        else:
            description = lambda: "'"+what+"' "
            ok = tok == what
        if not ok:
            sys.exit("testcrypt-func.h:{:d}: expected {}{}".format(
                pos, description(), why))
        return tok

    while True:
        tok = expect({"FUNC", "FUNC_WRAPPED"},
                     "at start of function specification", eof_ok=True)
        if tok is None:
            break

        expect("(", "after FUNC")
        rettype = expect(is_id, "return type")
        expect(",", "after return type")
        funcname = expect(is_id, "function name")
        expect(",", "after function name")
        args = []
        firstargkind = expect({"ARG", "VOID"}, "at start of argument list")
        if firstargkind == "VOID":
            expect(")", "after VOID")
        else:
            while True:
                # Every time we come back to the top of this loop, we've
                # just seen 'ARG'
                expect("(", "after ARG")
                argtype = expect(is_id, "argument type")
                expect(",", "after argument type")
                argname = expect(is_id, "argument name")
                args.append((argtype, argname))
                expect(")", "at end of ARG")
                punct = expect({",", ")"}, "after argument")
                if punct == ")":
                    break
                expect("ARG", "to begin next argument")
        yield funcname, rettype, args

def _setup(scope):
    valprefix = "val_"
    outprefix = "out_"
    optprefix = "opt_"
    consprefix = "consumed_"

    def trim_argtype(arg):
        if arg.startswith(optprefix):
            return optprefix + trim_argtype(arg[len(optprefix):])

        if (arg.startswith(valprefix) and
            "_" in arg[len(valprefix):]):
            # Strip suffixes like val_string_asciz
            arg = arg[:arg.index("_", len(valprefix))]
        return arg

    with open(os.path.join(putty_srcdir, "test", "testcrypt-func.h")) as f:
        header = f.read()
    tokens = _lex_testcrypt_header(header)
    for function, rettype, arglist in _parse_testcrypt_header(tokens):
        rettypes = []
        retnames = []
        if rettype != "void":
            rettypes.append(trim_argtype(rettype))
            retnames.append(None)

        argtypes = []
        argnames = []
        argsconsumed = []
        for arg, argname in arglist:
            if arg.startswith(outprefix):
                rettypes.append(trim_argtype(arg[len(outprefix):]))
                retnames.append(argname)
            else:
                consumed = False
                if arg.startswith(consprefix):
                    arg = arg[len(consprefix):]
                    consumed = True
                arg = trim_argtype(arg)
                argtypes.append((arg, consumed))
                argnames.append(argname)
        func = Function(function, rettypes, retnames,
                        argtypes, argnames)
        scope[function] = func
        if len(argtypes) > 0:
            t = argtypes[0][0]
            if t in method_prefixes:
                for prefix in method_prefixes[t]:
                    if function.startswith(prefix):
                        methodname = function[len(prefix):]
                        method_lists[t].append((methodname, func))
                        break

_setup(globals())
del _setup