path: root/cyberflex-shell.py

#!/usr/bin/env python
# -*- coding: iso-8859-1 -*-

import pycsc, crypto_utils, utils, cards, os, re, binascii, sys, exceptions, traceback, getopt, datetime
from shell import Shell

def list_readers():
    for index, name in enumerate(pycsc.listReader()):
        print "%i: %s" % (index, name)

class Logger(object):
    def __init__(self, filename, stream, prefix = "# "):
        self.fp = file(filename, "w")
        self.stream = stream
        self.prefix = prefix
        self.need_prefix = True
    
    def println(self, string):
        if not self.need_prefix:
            self.fp.write("\n")
            self.need_prefix = True
        self.fp.write("\n".join(string.splitlines()) + "\n")
    
    def flush(self):
        return self.stream.flush()
    
    def close(self):
        self.fp.close()
    
    def writelines(self, lines):
        for line in lines:
            self.write(line)
    
    def write(self, line):
        if self.need_prefix:
            self.fp.write(self.prefix)
            self.need_prefix = False
        
        self.fp.write( ( ("\n"+self.prefix).join(line.splitlines()) ) )
        if len(line) > 0 and line[-1] == "\n":
            self.fp.write("\n")
            self.need_prefix = True
        
        self.stream.write(line)

class Cyberflex_Shell(Shell):
    def __init__(self, basename):
        self.print_backtrace = True
        self.reader = 0
        self.logger = None
        Shell.__init__(self, basename)
        self.register_commands(self, self.NOCARD_COMMANDS)
        self.set_prompt("(No card) ")
    
    def cmd_runscript(self, filename, ask = True):
        "Run an APDU script from a file"
        fh = file(filename)
        
        doit = not ask
        ignored_SWs = []
        
        for line in fh:
            if line[:2] == "//" or line[:1] == "#":
                continue
            
            if not doit:
                print "?? %s" % line.strip()
                print "Execute? (Yes/No/All/Exit) ",
                answer = sys.stdin.readline()
                if answer[0].lower() in ('y', "\n"):
                    pass
                elif answer[0].lower() == 'n':
                    continue
                elif answer[0].lower() == 'a':
                    doit = True
                elif answer[0].lower() == 'e':
                    return
                else:
                    continue
            
            self.parse_and_execute(line)
            
            if self.card.sw_changed and self.card.last_sw != self.card.SW_OK \
                    and self.card.last_sw not in ignored_SWs:
                
                print "SW was not %s. Ignore (i) or Abort (a)? " % binascii.hexlify(self.card.SW_OK),
                answer = sys.stdin.readline()
                if answer[0].lower() in ('i', "\n"):
                    pass
                elif answer[0].lower() == 'a':
                    return
                elif answer[0] == 'S':
                    ignored_SWs.append(self.card.last_sw)
                    pass
                else:
                    return
    
    def cmd_listreaders(self):
        "List the available readers"
        list_readers()
    
    def cmd_enc(self, *args):
        "Encrypt or decrypt with openssl-like interface"
        
        args = list(args)
        
        MODE_DECRYPT = "-d"
        MODE_ENCRYPT = "-e"
        mode = MODE_ENCRYPT
        if "-d" in args:
            mode = MODE_DECRYPT
        
        input = None
        if "-in" in args:
            i = args.index("-in")
            input = args[i+1]
        
        if "-K" not in args:
            raise ValueError, "Must specify key with -K"
        i = args.index("-K")
        key = args[i+1]
        key = binascii.a2b_hex("".join(key.split()))
        
        iv = None
        if "-iv" in args:
            i = args.index("-iv")
            iv = args[i+1]
            iv = binascii.a2b_hex("".join(iv.split()))
        
        cipher = "des"
        if args[0][0] != "-":
            cipher = args[0]
        
        text = None
        if "-text" in args:
            if input is not None:
                raise ValueError, "Can't give -in and -text"
            i = args.index("-text")
            text = binascii.a2b_hex("".join(args[i+1].split()))
        
        if text is None:
            if input is None:
                text = self.card.last_result.data
            else:
                fp = file(input)
                text = fp.read()
                fp.close()
        
        result = crypto_utils.cipher(mode == MODE_ENCRYPT, cipher, key, text, iv)
        
        self.card.last_result = utils.R_APDU(result+"\x00\x00")
        print utils.hexdump(result)
    
    
    def cmd_eval(self, *args):
        "Execute raw python code"
        eval(" ".join(args))
        print

    def cmd_atr(self, *args):
        """Print the ATR of the currently inserted card."""
        print "ATR: %s" % utils.hexdump(self.card.card.status()['ATR'], short=True)
    
    def cmd_save_response(self, file_name, start = None, end = None):
        "Save the data in the last response to a file. start and end are optional"
        lastlen = len(self.card.last_result.data)
        if start is not None:
            start = (lastlen + (int(start,0) % lastlen) ) % lastlen
        else:
            start = 0
        if end is not None:
            end = (lastlen + (int(end,0) % lastlen) ) % lastlen
        else:
            end = lastlen
        
        fp = file(file_name, "w")
        try:
            fp.write(self.card.last_result.data[start:end])
        finally:
            fp.close()
    
    def cmd_disconnect(self, *args):
        "Close the connection to the currently inserted card"
        self.unregister_post_hook(self._print_sw)
        self.fallback = None
        self.unregister_pre_hook(self._clear_sw)
        self.unregister_pre_hook(self._update_prompt)
        self.unregister_commands(self.card)
        self.unregister_commands(self, self.CARD_COMMANDS)
        self.register_commands(self, self.NOCARD_COMMANDS)
        self.card.close_card()
        self.set_prompt("(No card) ")
    
    def cmd_reconnect(self, reader = None):
        "Re-open the connection to the card"
        self.cmd_disconnect()
        self.cmd_connect(reader)
    
    def cmd_fancy(self, *args):
        "Parse a fancy APDU and print the result"
        apdu = self.parse_fancy_apdu(*args)
        data = apdu.render()
        self.card.last_result = utils.R_APDU(data+"\x00\x00")
        print utils.hexdump(data)
    
    def _update_prompt(self):
        self.set_prompt(self.card.get_prompt() + " ")

    def _clear_sw(self):
        self.card.sw_changed = False
    
    _fancyapduregex = re.compile(r'^\s*([0-9a-f]{2}\s*){4,}\s*((xx|yy)\s*)?(([0-9a-f]{2}|:|\)|\(|\[|\])\s*)*$', re.I)
    @staticmethod
    def parse_fancy_apdu(*args):
        apdu_string = " ".join(args)
        if not Cyberflex_Shell._fancyapduregex.match(apdu_string):
            raise ValueError
        
        apdu_string = apdu_string.lower()
        have_le = False
        pos = apdu_string.find("xx")
        if pos == -1:
            pos = apdu_string.find("yy")
            have_le = True
        
        apdu_head = ""
        apdu_tail = apdu_string
        if pos != -1:
            apdu_head = apdu_string[:pos]
            apdu_tail = apdu_string[pos+2:]
        
        if apdu_head.strip() != "" and not Cyberflex_Shell._apduregex.match(apdu_head):
            raise ValueError
        
        class Node(list):
            def __init__(self, parent = None, type = None):
                list.__init__(self)
                self.parent = parent
                self.type = type
            
            def make_binary(self):
                "Recursively transform hex strings to binary"
                for index, child in enumerate(self):
                    if isinstance(child,str):
                        child = "".join( ("".join(child.split())).split(":") )
                        assert len(child) % 2 == 0
                        self[index] = binascii.a2b_hex(child)
                    else:
                        child.make_binary()
            
            def calculate_lengths(self):
                "Recursively calculate lengths and insert length counts"
                self.length = 0
                index = 0
                while index < len(self): ## Can't use enumerate() due to the insert() below
                    child = self[index]
                    
                    if isinstance(child,str):
                        self.length = self.length + len(child)
                    else:
                        child.calculate_lengths()
                        
                        formatted_len = binascii.a2b_hex("%02x" % child.length) ## FIXME len > 255?
                        self.length = self.length + len(formatted_len) + child.length
                        self.insert(index, formatted_len)
                        index = index + 1
                    
                    index = index + 1
            
            def flatten(self, offset = 0, ignore_types=["("]):
                "Recursively flatten, gather list of marks"
                string_result = []
                mark_result = []
                for child in self:
                    if isinstance(child,str):
                        string_result.append(child)
                        offset = offset + len(child)
                    else:
                        start = offset
                        child_string, child_mark = child.flatten(offset, ignore_types)
                        string_result.append(child_string)
                        offset = end = offset + len(child_string)
                        if not child.type in ignore_types:
                            mark_result.append( (child.type, start, end) )
                        mark_result.extend(child_mark)
                
                return "".join(string_result), mark_result
        
        
        tree = Node()
        current = tree
        allowed_parens = {"(": ")", "[":"]"}
        
        for pos,char in enumerate(apdu_tail):
            if char in (" ", "a", "b", "c", "d", "e", "f",":") or char.isdigit():
                if len(current) > 0 and isinstance(current[-1],str):
                    current[-1] = current[-1] + char
                else:
                    current.append(str(char))
                
            elif char in allowed_parens.values():
                if current.parent is None:
                    raise ValueError
                if allowed_parens[current.type] != char:
                    raise ValueError
                
                current = current.parent
                
            elif char in allowed_parens.keys():
                current.append( Node(current, char) )
                current = current[-1]
                
            else:
                raise ValueError
        
        if current != tree:
            raise ValueError
        
        tree.make_binary()
        tree.calculate_lengths()
        
        apdu_head = apdu_head.strip()
        if apdu_head != "":
            l = tree.length
            if have_le: 
                l = l - 1 ## FIXME Le > 255?
            formatted_len = "%02x" % l  ## FIXME len > 255?
            apdu_head = binascii.a2b_hex("".join( (apdu_head + formatted_len).split() ))
        
        apdu_tail, marks = tree.flatten(offset=0)
        
        apdu = utils.C_APDU(apdu_head + apdu_tail, marks = marks)
        return apdu
    
    def do_fancy_apdu(self, *args):
        apdu = None
        try:
            apdu = Cyberflex_Shell.parse_fancy_apdu(*args)
        except ValueError:
            raise NotImplementedError
        
        if apdu is not None:
            return self.do_apdu(apdu)
    
    _apduregex = re.compile(r'^\s*([0-9a-f]{2}\s*){4,}$', re.I)
    def do_raw_apdu(self, *args):
        apdu_string = "".join(args)
        if not Cyberflex_Shell._apduregex.match(apdu_string):
            raise NotImplementedError
        
        apdu_binary = binascii.a2b_hex("".join(apdu_string.split()))
        apdu = utils.C_APDU(apdu_binary)
        
        return self.do_apdu(apdu)
    
    def do_apdu(self, apdu):
        response = self.card.send_apdu(apdu)
        
        if len(response.data) > 0: ## The SW is already printed by _print_sw as a post_hook
            print utils.hexdump(response.data)
    
    def pause_log(self):
        if self.logger is not None:
            sys.stdout = self.logger.stream
    
    def unpause_log(self):
        if self.logger is not None:
            sys.stdout = self.logger
    
    def start_log(self, filename):
        if self.logger is not None:
            self.stop_log()
        self.logger = Logger(filename, sys.stdout)
        sys.stdout = self.logger
        print "Logging to %s" % filename
        self.register_pre_hook(self.pause_log)
    
    def stop_log(self):
        if self.logger is not None:
            print "Log stopped"
            sys.stdout = self.logger.stream
            self.logger.flush()
            self.logger = None
            self.unregister_pre_hook(self.pause_log)
    
    def parse_and_execute(self, line):
        if self.logger is not None:
            self.logger.println( self.logger.prefix + "\n" 
                + self.logger.prefix + "=== " + datetime.datetime.now().isoformat(" ") + " " + ("="*49) )
            self.logger.println(line)
        if self.logger is not None:
            self.unpause_log()
        result = Shell.parse_and_execute(self, line)
        return result
    
    def cmd_log(self, filename = None):
        "Start (when given a filename) or stop (otherwise) logging to a file"
        if filename is not None:
            date = datetime.datetime.now()
            vars = {
                "HOMEDIR": os.environ["HOME"],
                "ISOTIME": date.isoformat()
            }
            self.start_log(filename % vars)
        else:
            self.stop_log()

    def _print_sw(self):
        if self.card.sw_changed:
            print self.card.decode_statusword()
    
    def _find_driver_class(driver_name):
        for i in dir(cards):
            _obj = getattr(cards, i)
            if driver_name.lower() == i.lower():
                return _obj
            if hasattr(_obj, "DRIVER_NAME") and driver_name.lower() == getattr(_obj, "DRIVER_NAME").lower():
                return _obj
        raise NameError, "Class not found"
    
    _find_driver_class = staticmethod(_find_driver_class)
    
    def cmd_unloaddriver(self, driver_name):
        "Remove a driver from the current connection"
        self.unregister_commands(self.card)
        try:
            self.card.remove_classes( [self._find_driver_class(driver_name)] )
        finally:
            self.register_commands(self.card)
    
    def cmd_loaddriver(self, driver_name):
        "Add a driver to the current connection"
        self.unregister_commands(self.card)
        try:
            self.card.add_classes( [self._find_driver_class(driver_name)] )
        finally:
            self.register_commands(self.card)
    
    def cmd_connect(self, reader = None):
        "Open the connection to a card"
        if reader is None:
            reader = self.reader
        
        if isinstance(reader, int) or reader.isdigit():
            reader = int(reader)
            readerName = pycsc.listReader()[reader]
        else:
            readerName = reader
        self.reader = reader
        
        newState = pycsc.getStatusChange(ReaderStates=[
                {'Reader': readerName, 'CurrentState':pycsc.SCARD_STATE_UNAWARE}
            ]
        )
        
        print "Using reader: %s" % readerName
        print "Card present: %s" % ((newState[0]['EventState'] & pycsc.SCARD_STATE_PRESENT) and "yes" or "no")
        
        if not newState[0]['EventState'] & pycsc.SCARD_STATE_PRESENT:
            print "Please insert card ..."
            
            last_was_mute = False
            
            while not newState[0]['EventState'] & pycsc.SCARD_STATE_PRESENT \
                or newState[0]['EventState'] & pycsc.SCARD_STATE_MUTE:
                
                try:
                    newState = pycsc.getStatusChange(ReaderStates=[
                            {'Reader': readerName, 'CurrentState':newState[0]['EventState']}
                        ], Timeout = 100 
                    ) ## 100 ms latency from Ctrl-C to abort should be almost unnoticeable by the user
                except pycsc.PycscException, e:
                    if e.args[0] == 'Command timeout.': pass ## ugly
                    else: raise
                
                if newState[0]['EventState'] & pycsc.SCARD_STATE_MUTE:
                    if not last_was_mute:
                        print "Card is mute, please retry ..."
                    last_was_mute = True
                else: 
                    last_was_mute = False
                
            print "Card present: %s" % ((newState[0]['EventState'] & pycsc.SCARD_STATE_PRESENT) and "yes" or "no")
        
        print "ATR:          %s" % utils.hexdump(newState[0]['Atr'], short = True)
        
        pycsc_card = pycsc.pycsc(reader = readerName, protocol = pycsc.SCARD_PROTOCOL_ANY)
        self.card = cards.new_card_object(pycsc_card)
        
        self.unregister_commands(self, self.NOCARD_COMMANDS)
        self.register_commands(self, self.CARD_COMMANDS)
        self.register_commands(self.card)
        
        self.register_pre_hook(self._update_prompt)
        self.register_pre_hook(self._clear_sw)
        
        shell.fallback = self.do_fancy_apdu
        
        shell.register_post_hook(self._print_sw)
    
    COMMANDS = dict(Shell.COMMANDS)
    COMMANDS.update( {
        "list_readers": cmd_listreaders,
        "eval": cmd_eval,
        "save_response": cmd_save_response,
        "fancy": cmd_fancy,
        "enc": cmd_enc,
        "log": cmd_log,
    } )
    
    CARD_COMMANDS = {
        "atr": cmd_atr,
        "disconnect": cmd_disconnect,
        "reconnect": cmd_reconnect,
        "driver_load": cmd_loaddriver,
        "driver_unload": cmd_unloaddriver,
        "run_script": cmd_runscript,
    }
    
    NOCARD_COMMANDS = {
        "connect": cmd_connect,
    }

def usage():
    print """Cyberflex shell
Synopsis: cyberflex-shell.py [options] [scriptfiles]
Options:
    -r, --reader             Select the reader to use, either by
                             index or by name
    -l, --list-readers       List the available readers and their
                             indices
    -n, --dont-connect       Don't connect to the card on startup
    -y, --dont-ask           Don't ask for confirmation for every
                             command run from the scriptfiles
    -i, --force-interactive  Force interactive mode after running
                             scripts from the command line
    -h, --help               This help
"""

OPTIONS = "r:lnyih"
LONG_OPTIONS = ["reader=", "list-readers","dont-connect","dont-ask","force-interactive","help"]
exit_now = False
dont_connect = False
dont_ask = False
force_interactive = False
reader = None

if __name__ == "__main__":
    
    (options, arguments) = getopt.gnu_getopt(sys.argv[1:], OPTIONS, LONG_OPTIONS)
    
    for (option, value) in options:
        if option in ("-r","--reader"):
            reader = value
        if option in ("-l","--list-readers"):
            list_readers()
            exit_now = True
        if option in ("-h","--help"):
            usage()
            exit_now = True
        if option in ("-n","--dont-connect"):
            dont_connect = True
        if option in ("-y","--dont-ask"):
            dont_ask = True
        if option in ("-i","--force-interactive"):
            force_interactive = True
    
    if exit_now:
        sys.exit()
    del exit_now
    
    print "Cyberflex shell"
    shell = Cyberflex_Shell("cyberflex-shell")
    
    if not dont_connect:
        shell.cmd_connect(reader)
    
    shell.run_startup()
    
    for filename in arguments:
        shell.cmd_runscript(filename, not dont_ask)
    
    if len(arguments) == 0 or force_interactive:
        shell.run()