Lint fixes

libagent/device/keepkey.py

@@ -32,11 +32,12 @@ def _defs(self):
 
     required_version = '>=1.0.4'
 
-    def pubkey(self, identity, ecdh=False):
+    def pubkey(self, identity, ecdh=False, options=None):
         """Return public key."""
         _verify_support(identity, ecdh)
-        return trezor.Trezor.pubkey(self, identity=identity, ecdh=ecdh)
+        return trezor.Trezor.pubkey(self, identity=identity, ecdh=ecdh,
+                                    options=options)
 
-    def ecdh(self, identity, pubkey):
+    def ecdh(self, identity, pubkey, options=None):
         """No support for ECDH in KeepKey firmware."""
         _verify_support(identity, ecdh=True)

libagent/device/ledger.py

@@ -41,10 +41,6 @@ def package_name(cls):
         """Python package name (at PyPI)."""
         return 'ledger-agent'
 
-    def __init__(self, config=None):
-        self.config = config
-        super(LedgerNanoS, self).__init__(config=config)
-
     def connect(self):
         """Enumerate and connect to the first USB HID interface."""
         try:

libagent/device/trezor.py

@@ -5,7 +5,6 @@
 import os
 import subprocess
 import sys
-import traceback
 
 import mnemonic
 import semver
@@ -15,10 +14,6 @@
 log = logging.getLogger(__name__)
 
 
-class UserCancelException(RuntimeError):
-    pass
-
-
 def _message_box(label, sp=subprocess):
     """Launch an external process for PIN/passphrase entry GUI."""
     cmd = ('import sys, pymsgbox; '
@@ -37,14 +32,14 @@ def _is_open_tty(stream):
     return not stream.closed and os.isatty(stream.fileno())
 
 
-def _pin_communicate(self, program, message, error=None, options={}):
+def _pin_communicate(program, message, error=None, options=None):
     args = [program]
+    options = options or {}
     if 'DISPLAY' in os.environ:
         args.extend(['--display', os.environ['DISPLAY']])
     try:
         entry = subprocess.Popen(
             args,
-            encoding='utf-8',
             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
         )
     except OSError as e:
@@ -54,19 +49,19 @@ def _pin_communicate(self, program, message, error=None, options={}):
             raise
 
     def expect(prefix=None):
-        line = entry.stdout.readline()
+        line = entry.stdout.readline().decode('utf-8')
         if line.endswith('\n'):
             line = line[:-1]
-        log.debug('PINENTRY <- {}'.format(line))
+        log.debug('PINENTRY <- %s', line)
         if line.startswith('ERR '):
-            raise UserCancelException()
+            raise RuntimeError(line)
         if prefix and not line.startswith(prefix):
             raise RuntimeError('Received unexpected response from pinentry')
         return line[len(prefix) if prefix else 0:]
 
     def send(line):
-        log.debug('PINENTRY -> {}'.format(line))
-        entry.stdin.write('{}\n'.format(line))
+        log.debug('PINENTRY -> %s', line)
+        entry.stdin.write('{}\n'.format(line).encode('utf-8'))
         entry.stdin.flush()
 
     expect('OK')
@@ -87,7 +82,7 @@ def send(line):
     entry.stdin.close()
     try:
         entry.communicate()
-    except:
+    except:  # pylint: disable=bare-except
         pass
     return result
 
@@ -101,6 +96,7 @@ def package_name(cls):
         return 'trezor-agent'
 
     def __init__(self, config=None):
+        """Set up Trezor device object."""
         self.config = config or {}
         self.options = {}
         super(Trezor, self).__init__(config=config)
@@ -132,7 +128,6 @@ def new_handler(msg):
             result = None
             pinentry_program = self.config.get('pinentry-program')
             scrambled_pin = _pin_communicate(
-                self,
                 pinentry_program or 'pinentry',
                 'Please enter your Trezor PIN' if pinentry_program
                 else fallback_message,
@@ -166,7 +161,6 @@ def new_handler(msg):
             ack = None
             passphrase_program = self.config.get('passentry-program')
             passphrase = _pin_communicate(
-                self,
                 passphrase_program or 'pinentry',
                 'Please enter your passphrase',
                 options=self.options,
@@ -218,9 +212,6 @@ def connect(self):
                 except (self._defs.PinException, ValueError) as e:
                     log.error('Invalid PIN: %s, retrying...', e)
                     continue
-                except UserCancelException:
-                    connection.cancel()
-                    return None
                 except Exception as e:
                     log.exception('ping failed: %s', e)
                     connection.close()  # so the next HID open() will succeed

libagent/gpg/agent.py

@@ -111,6 +111,7 @@ def __init__(self, device, pubkey_bytes):
         }
 
     def set_option(self, args):
+        """Store options passed to the agent."""
         parts = args[0].decode('utf-8').split('=', 1)
         i = iter(parts)
         k = next(i, None)

libagent/ssh/tests/test_client.py

@@ -20,11 +20,11 @@ class MockDevice(device.interface.Device):  # pylint: disable=abstract-method
     def connect(self):  # pylint: disable=no-self-use
         return mock.Mock()
 
-    def pubkey(self, identity, ecdh=False):  # pylint: disable=unused-argument
+    def pubkey(self, identity, ecdh=False, options=None):  # pylint: disable=unused-argument, line-too-long
         assert self.conn
         return PUBKEY
 
-    def sign(self, identity, blob):
+    def sign(self, identity, blob, options=None):
         """Sign given blob and return the signature (as bytes)."""
         assert self.conn
         assert blob == BLOB