cli.py

# -*- coding: latin-1 -*-
# -----------------------------------------------------------------------------
# Copyright 2012, 2017 Stephen Tiedemann <stephen.tiedemann@gmail.com>
#
# Licensed under the EUPL, Version 1.1 or - as soon they
# will be approved by the European Commission - subsequent
# versions of the EUPL (the "Licence");
# You may not use this work except in compliance with the
# Licence.
# You may obtain a copy of the Licence at:
#
# https://joinup.ec.europa.eu/software/page/eupl
#
# Unless required by applicable law or agreed to in
# writing, software distributed under the Licence is
# distributed on an "AS IS" basis,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
# express or implied.
# See the Licence for the specific language governing
# permissions and limitations under the Licence.
# -----------------------------------------------------------------------------
import logging
import re
import time
import errno
import inspect
import threading
from operator import itemgetter
import platform
import nfc


log = logging.getLogger('main')


def get_test_methods(obj):
    test_methods = list()
    for name, func in inspect.getmembers(obj, inspect.ismethod):
        if name.startswith("test_"):
            line = inspect.getsourcelines(func)[1]
            text = inspect.getdoc(func)
            test_methods.append((line, name.lstrip("test_"), text))
    return test_methods


class TestFail(Exception):
    def __init__(self, value):
        self.value = value

    def __str__(self):
        return str(self.value)


class TestSkip(Exception):
    def __init__(self, value):
        self.value = value

    def __str__(self):
        return str(self.value)


class CommandLineInterface(object):
    def __init__(self, argument_parser, groups=''):
        self.groups = groups.split()
        self.test_completed = False
        for group in self.groups:
            eval("self.add_{0}_options".format(group))(argument_parser)

        argument_parser.add_argument(
                "-l", "--loop", action="store_true",
                help="restart after termination")

        self.options = argument_parser.parse_args()

        lvl = logging.ERROR if self.options.quiet else logging.INFO
        if self.options.debug and not self.options.logfile:
            lvl = logging.DEBUG - (1 if self.options.verbose else 0)

        fmt = '[%(name)s] %(message)s'
        if self.options.reltime:
            fmt = '%(relativeCreated)d ms ' + fmt
        if self.options.abstime:
            fmt = '%(asctime)s ' + fmt

        ch = ColorStreamHandler()
        ch.setLevel(lvl)
        ch.setFormatter(logging.Formatter(fmt))
        logging.getLogger().addHandler(ch)

        if self.options.logfile:
            fmt = '%(asctime)s [%(name)s] %(message)s'
            fh = logging.FileHandler(self.options.logfile, "w")
            fh.setFormatter(logging.Formatter(fmt))
            fh.setLevel(logging.DEBUG - (1 if self.options.verbose else 0))
            logging.getLogger().addHandler(fh)

        logging.getLogger().setLevel(logging.NOTSET)
        logging.getLogger('main').setLevel(logging.INFO)
        for module in self.options.debug:
            log.info("enable debug output for '{0}'".format(module))
            logging.getLogger(module).setLevel(1)

        log.debug(self.options)

        if "test" in self.groups:
            if self.options.test_all:
                # get_test_method() yields (line, name, docstr), ..., ...
                test_methods = sorted(get_test_methods(self),
                                      key=itemgetter(0))
                self.options.test = list(map(itemgetter(1), test_methods))

            if len(self.options.test) > 0 and self.options.select:
                self.options.test = filter(
                    lambda name: re.match(self.options.select, name),
                    self.options.test)

    def add_dbg_options(self, argument_parser):
        group = argument_parser.add_argument_group(title="Debug Options")
        group.add_argument(
                "-d", metavar="MODULE", dest="debug", action="append",
                default=list(),
                help="enable debug log for MODULE (main, nfc.clf, ...)")
        group.add_argument(
                "-v", "--verbose", action="store_true",
                help="show more information")
        group.add_argument(
                "-q", "--quiet", action="store_true",
                help="show less information")
        group.add_argument(
                "-f", dest="logfile", metavar="LOGFILE",
                help="write debug logs to LOGFILE (with date and time)")
        group.add_argument(
                "--reltime", action="store_true",
                help="show relative timestamps in screen log")
        group.add_argument(
                "--abstime", action="store_true",
                help="show absolute timestamps in screen log")

    def add_llcp_options(self, argument_parser):
        group = argument_parser.add_argument_group(title="Peer Mode Options")
        group.add_argument(
                "--miu", type=int, default=2175, metavar='',
                help="LLC Link MIU octets (default: %(default)s octets)")
        group.add_argument(
                "--lto", type=int, default=500, metavar='',
                help="LLC Link Timeout in ms (default: %(default)s ms)")
        group.add_argument(
                "--lsc", type=int, choices=range(3), default=3, metavar='',
                help="LLC Link Service Class (default: %(default)s)")
        group.add_argument(
                "--rwt", type=int, default=8, metavar='',
                help="DEP Response Waiting Time index (default: %(default)s)")
        group.add_argument(
                "--mode", choices=["t", "target", "i", "initiator"],
                metavar='',
                help="connect as [t]arget or [i]nitiator (default: both)")
        group.add_argument(
                "--bitrate", type=int, default=424, metavar='',
                choices=(106, 212, 424), help="""\
                DEP Initiator bitrate 106/212/424 (default: %(default)s)""")
        group.add_argument(
                "--passive-only", action="store_true",
                help="only passive mode activation when initiator")
        group.add_argument(
                "--listen-time", type=int, default=250, metavar='',
                help="DEP Target listen time in ms (default: %(default)s ms)")
        group.add_argument(
                "--no-aggregation", action="store_true",
                help="disable outbound packet aggregation")
        group.add_argument(
                "--no-encryption", action="store_true",
                help="disable secure data transport")

    def add_rdwr_options(self, argument_parser):
        group = argument_parser.add_argument_group(title="Reader Mode Options")
        group.add_argument(
                "--wait", action="store_true",
                help="wait until tag removed (implicit with '-l')")
        group.add_argument(
                "--technology", choices=list("ABFabf"), metavar="{A,B,F}",
                help="poll for a single technology (default: all)")

    def add_card_options(self, argument_parser):
        argument_parser.add_argument_group(title="Card Mode Options")

    def add_clf_options(self, argument_parser):
        group = argument_parser.add_argument_group(title="Device Options")
        group.add_argument(
                "--device", metavar="PATH", action="append", help="""
                use contactless reader at:
                'usb[:vid[:pid]]' (with vendor and product id),
                'usb[:bus[:dev]]' (with bus and device number),
                'tty:port:driver' (with /dev/tty<port> and <driver>),
                'com:port:driver' (with COM<port> and <driver>),
                'udp[:host[:port]]' (with <host> name/addr and <port> number)
                """)

    def add_iop_options(self, argument_parser):
        group = argument_parser.add_argument_group(
                title="Interoperability Options")
        group.add_argument(
                "--quirks", action="store_true",
                help="support non-compliant implementations")

    def add_test_options(self, argument_parser):
        group = argument_parser.add_argument_group(
                title="Test options")
        group.add_argument(
                "-t", "--test", default=[], action="append",
                metavar="T", help="add test name <T> to test schedule")
        group.add_argument(
                "-T", "--test-all", action="store_true",
                help="add all available tests to schedule")
        group.add_argument(
                "--select", metavar="REGEX",
                help="from schedule select tests matching REGEX")

        test_name_and_text, max_name_length = list(), 0
        for line, name, text in sorted(get_test_methods(self),
                                       key=itemgetter(0)):
            test_name_and_text.append((name, text.splitlines()[0]))
            max_name_length = max(max_name_length, len(name))

        argument_parser.description += "\nAvailable Tests:\n"
        for name, text in test_name_and_text:
            argument_parser.description += '  {0}   {1}\n'.format(
                    name.ljust(max_name_length), text)

    def on_rdwr_startup(self, targets):
        return targets

    def on_rdwr_connect(self, tag):
        log.info(tag)
        return True

    def on_llcp_startup(self, llc):
        if "test" in self.groups and len(self.options.test) == 0:
            log.error("no test specified")
            return None
        return llc

    def on_llcp_connect(self, llc):
        if "test" in self.groups:
            self.test_completed = False
            threading.Thread(target=self.run_tests, args=(llc,)).start()
            llc.run(terminate=self.terminate)
            return False
        return True

    def on_card_startup(self, target):
        log.warning("on_card_startup should be customized")
        return None

    def on_card_connect(self, tag):
        log.info("activated as {0}".format(tag))
        if "test" in self.groups:
            self.test_completed = False
            self.run_tests(tag)
            return False
        return True

    def on_card_release(self, tag):
        return True

    def terminate(self):
        return self.test_completed

    def run_tests(self, *args):
        if len(self.options.test) > 1:
            log.info("run tests: {0}".format(self.options.test))
        for index, test in enumerate(self.options.test):
            test_name = "test_{0}".format(test)
            try:
                test_func = eval("self." + test_name)
            except AttributeError:
                log.error("invalid test '{0}'".format(test))
                continue
            test_info = test_func.__doc__.splitlines()[0]
            try:
                test_name = "Test {0:02d}".format(test)
            except ValueError:
                test_name = test
            print("{0}: {1}".format(test_name, test_info))
            try:
                test_func(*args)
            except (TestFail, AssertionError) as error:
                print("{0}: FAIL ({1})".format(test_name, error))
            except TestSkip as error:
                print("{0}: SKIP ({1})".format(test_name, error))
            else:
                print("{0}: PASS".format(test_name))
            if index < len(self.options.test) - 1:
                time.sleep(1)
        self.test_completed = True

    def run_once(self):
        if self.options.device is None:
            self.options.device = ['usb']

        for path in self.options.device:
            try:
                clf = nfc.ContactlessFrontend(path)
            except IOError as error:
                if error.errno == errno.ENODEV:
                    log.info("no contactless reader found on " + path)
                elif error.errno == errno.EACCES:
                    log.info("access denied for device with path " + path)
                elif error.errno == errno.EBUSY:
                    log.info("the reader on " + path + " is busy")
                else:
                    log.debug(repr(error) + "when trying " + path)
            else:
                log.debug("found a usable reader on " + path)
                break
        else:
            log.error("no contactless reader available")
            raise SystemExit(1)

        if "rdwr" in self.groups:
            rdwr_options = {
                'on-startup': self.on_rdwr_startup,
                'on-connect': self.on_rdwr_connect,
            }
            if self.options.technology:
                rdwr_options["targets"] = {
                    "A": ["106A"],
                    "B": ["106B"],
                    "F": ["212F"],
                }[self.options.technology.upper()]
        else:
            rdwr_options = None

        if "llcp" in self.groups:
            if self.options.mode is None:
                self.options.role = None
            elif self.options.mode in ('t', 'target'):
                self.options.role = 'target'
            elif self.options.mode in ('i', 'initiator'):
                self.options.role = 'initiator'
            llcp_options = {
                'on-startup': self.on_llcp_startup,
                'on-connect': self.on_llcp_connect,
                'role':       self.options.role,
                'brs':        (106, 212, 424).index(self.options.bitrate),
                'acm':        (not self.options.passive_only),
                'rwt':        self.options.rwt,
                'miu':        self.options.miu,
                'lto':        self.options.lto,
                'lsc':        self.options.lsc,
                'agf':        (not self.options.no_aggregation),
                'sec':        (not self.options.no_encryption),
            }
        else:
            llcp_options = None

        if "card" in self.groups:
            card_options = {
                'on-startup': self.on_card_startup,
                'on-connect': self.on_card_connect,
                'on-release': self.on_card_release,
                'targets':    [],
            }
        else:
            card_options = None

        try:
            kwargs = {'llcp': llcp_options,
                      'rdwr': rdwr_options,
                      'card': card_options}
            return clf.connect(**kwargs)
        finally:
            clf.close()

    def run(self):
        while self.run_once() and self.options.loop:
            log.info("*** RESTART ***")


# ColorStreamHandler for python logging framework.
# based on: http://stackoverflow.com/questions/384076/1336640#1336640

# Copyright (c) 2014 Markus Pointner
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

class AnsiColorStreamHandler(logging.StreamHandler):
    DEFAULT = '\x1b[0m'
    RED = '\x1b[31m'
    GREEN = '\x1b[32m'
    YELLOW = '\x1b[33m'
    BLUE = '\x1b[34m'
    CYAN = '\x1b[36m'

    CRITICAL = RED
    ERROR = RED
    WARNING = YELLOW
    INFO = GREEN
    DEBUG = CYAN
    VERBOSE = BLUE

    @classmethod
    def _get_color(cls, level):
        if level >= logging.CRITICAL:
            return cls.CRITICAL
        elif level >= logging.ERROR:
            return cls.ERROR
        elif level >= logging.WARNING:
            return cls.WARNING
        elif level >= logging.INFO:
            return cls.INFO
        elif level >= logging.DEBUG:
            return cls.DEBUG
        elif level >= logging.DEBUG - 1:
            return cls.VERBOSE
        else:
            return cls.DEFAULT

    def format(self, record):
        text = logging.StreamHandler.format(self, record)
        color = self._get_color(record.levelno)
        return color + text + self.DEFAULT


class WindowsColorStreamHandler(logging.StreamHandler):
    # wincon.h
    FOREGROUND_BLACK = 0x0000
    FOREGROUND_BLUE = 0x0001
    FOREGROUND_GREEN = 0x0002
    FOREGROUND_CYAN = 0x0003
    FOREGROUND_RED = 0x0004
    FOREGROUND_MAGENTA = 0x0005
    FOREGROUND_YELLOW = 0x0006
    FOREGROUND_GREY = 0x0007
    FOREGROUND_INTENSITY = 0x0008  # foreground color is intensified.
    FOREGROUND_WHITE = FOREGROUND_BLUE | FOREGROUND_GREEN | FOREGROUND_RED

    BACKGROUND_BLACK = 0x0000
    BACKGROUND_BLUE = 0x0010
    BACKGROUND_GREEN = 0x0020
    BACKGROUND_CYAN = 0x0030
    BACKGROUND_RED = 0x0040
    BACKGROUND_MAGENTA = 0x0050
    BACKGROUND_YELLOW = 0x0060
    BACKGROUND_GREY = 0x0070
    BACKGROUND_INTENSITY = 0x0080  # background color is intensified.

    DEFAULT = FOREGROUND_WHITE
    CRITICAL = BACKGROUND_YELLOW | FOREGROUND_RED | FOREGROUND_INTENSITY \
        | BACKGROUND_INTENSITY
    ERROR = FOREGROUND_RED | FOREGROUND_INTENSITY
    WARNING = FOREGROUND_YELLOW | FOREGROUND_INTENSITY
    INFO = FOREGROUND_GREEN
    DEBUG = FOREGROUND_CYAN
    VERBOSE = FOREGROUND_BLUE

    @classmethod
    def _get_color(cls, level):
        if level >= logging.CRITICAL:
            return cls.CRITICAL
        elif level >= logging.ERROR:
            return cls.ERROR
        elif level >= logging.WARNING:
            return cls.WARNING
        elif level >= logging.INFO:
            return cls.INFO
        elif level >= logging.DEBUG:
            return cls.DEBUG
        elif level >= logging.DEBUG - 1:
            return cls.VERBOSE
        else:
            return cls.DEFAULT

    def _set_color(self, code):
        import ctypes
        ctypes.windll.kernel32.SetConsoleTextAttribute(self._outhdl, code)

    def __init__(self, stream=None):
        super(WindowsColorStreamHandler, self).__init__(stream)
        # get file handle for the stream
        import msvcrt
        self._outhdl = msvcrt.get_osfhandle(self.stream.fileno())

    def emit(self, record):
        color = self._get_color(record.levelno)
        self._set_color(color)
        logging.StreamHandler.emit(self, record)
        self._set_color(self.FOREGROUND_WHITE)


# select ColorStreamHandler based on platform
if platform.system() == 'Windows':
    ColorStreamHandler = WindowsColorStreamHandler
else:
    ColorStreamHandler = AnsiColorStreamHandler