Add installudevrules command for linux

hwi.spec

@@ -25,6 +25,10 @@ a = Analysis(['hwi.py'],
              win_private_assemblies=False,
              cipher=block_cipher,
              noarchive=False)
+
+if platform.system() == 'Linux':
+    a.datas += Tree('udev', prefix='udev')
+
 pyz = PYZ(a.pure, a.zipped_data,
              cipher=block_cipher)
 exe = EXE(pyz,

hwilib/cli.py

@@ -2,7 +2,7 @@
 
 from .commands import backup_device, displayaddress, enumerate, find_device, \
     get_client, getmasterxpub, getxpub, getkeypool, prompt_pin, restore_device, send_pin, setup_device, \
-    signmessage, signtx, wipe_device
+    signmessage, signtx, wipe_device, install_udev_rules
 from .errors import (
     HWWError,
     NO_DEVICE_PATH,
@@ -63,6 +63,9 @@ def prompt_pin_handler(args, client):
 def send_pin_handler(args, client):
     return send_pin(client, pin=args.pin)
 
+def install_udev_rules_handler(args):
+    return install_udev_rules(args.source, args.location)
+
 def process_commands(cli_args):
     parser = argparse.ArgumentParser(description='Hardware Wallet Interface, version {}.\nAccess and send commands to a hardware wallet device. Responses are in JSON format.'.format(__version__), formatter_class=argparse.RawDescriptionHelpFormatter)
     parser.add_argument('--device-path', '-d', help='Specify the device path of the device to connect to')
@@ -142,6 +145,13 @@ def process_commands(cli_args):
     sendpin_parser.add_argument('pin', help='The numeric positions of the PIN')
     sendpin_parser.set_defaults(func=send_pin_handler)
 
+    if sys.platform.startswith("linux"):
+        udevrules_parser = subparsers.add_parser('installudevrules', help='Install and load the udev rule files for the hardware wallet devices')
+        udevrules_parser.add_argument('--source', help=argparse.SUPPRESS, default='udev')
+        udevrules_parser.add_argument('--location', help='The path where the udev rules files will be copied', default='/etc/udev/rules.d/')
+        udevrules_parser.set_defaults(func=install_udev_rules_handler)
+
+
     if any(arg == '--stdin' for arg in cli_args):
         blank_count = 0
         while True:
@@ -177,6 +187,17 @@ def process_commands(cli_args):
     if command == 'enumerate':
         return args.func(args)
 
+    # Install the devices udev rules for Linux
+    if command == 'installudevrules':
+        try:
+            result = args.func(args)
+        except Exception as e:
+            if args.debug:
+                import traceback
+                traceback.print_exc()
+            result = {'error': str(e), 'code': UNKNOWN_ERROR}
+        return result
+
     # Auto detect if we are using fingerprint or type to identify device
     if args.fingerprint or (args.device_type and not args.device_path):
         client = find_device(args.device_path, args.password, args.device_type, args.fingerprint)

hwilib/commands.py

@@ -9,6 +9,8 @@ from .base58 import get_xpub_fingerprint_as_id, get_xpub_fingerprint_hex, xpub_t
 from .errors import NoPasswordError, UnavailableActionError, DeviceAlreadyInitError, DeviceAlreadyUnlockedError, UnknownDeviceError, BAD_ARGUMENT, NOT_IMPLEMENTED
 from .descriptor import Descriptor
 from .devices import __all__ as all_devs
+from .udevinstaller import UDevInstaller
+
 
 # Get the client for the device
 def get_client(device_type, device_path, password=''):
@@ -186,3 +188,6 @@ def prompt_pin(client):
 
 def send_pin(client, pin):
     return client.send_pin(pin)
+
+def install_udev_rules(source, location):
+    return UDevInstaller.install(source, location);

hwilib/errors.py

@@ -16,6 +16,7 @@ DEVICE_NOT_READY = -12
 UNKNOWN_ERROR = -13
 ACTION_CANCELED = -14
 DEVICE_BUSY = -15
+NEED_TO_BE_ROOT = -16
 
 # Exceptions
 class HWWError(Exception):

hwilib/udevinstaller.py

@@ -0,0 +1,60 @@
+import sys
+from subprocess import check_call, CalledProcessError, DEVNULL
+from .errors import NEED_TO_BE_ROOT
+from shutil import copy
+from os import path, environ, listdir, getlogin, geteuid
+
+class UDevInstaller(object):
+    @staticmethod
+    def install(source, location):
+        try:
+            udev_installer = UDevInstaller()
+            udev_installer.copy_udev_rule_files(source, location)
+            udev_installer.trigger()
+            udev_installer.reload_rules()
+            udev_installer.add_user_plugdev_group()
+        except CalledProcessError as e:
+            if geteuid() != 0:
+                return {'error':'Need to be root.','code':NEED_TO_BE_ROOT}
+            raise
+        return {"success": True}
+
+    def __init__(self):
+        self._udevadm = '/sbin/udevadm'
+        self._groupadd = '/usr/sbin/groupadd'
+        self._usermod = '/usr/sbin/usermod'
+
+    def _execute(self, command, *args):
+        command = [command] + list(args)
+        check_call(command, stderr=DEVNULL, stdout=DEVNULL)
+
+    def trigger(self):
+        self._execute(self._udevadm, 'trigger')
+
+    def reload_rules(self):
+        self._execute(self._udevadm, 'control', '--reload-rules')
+    
+    def add_user_plugdev_group(self):
+        self._create_group('plugdev')
+        self._add_user_to_group(getlogin(), 'plugdev')
+
+    def _create_group(self, name):
+        try:
+            self._execute(self._groupadd, name)
+        except CalledProcessError as e:
+            if e.returncode != 9: # group already exists
+                raise
+
+    def _add_user_to_group(self, user, group):
+        self._execute(self._usermod, '-aG', group, user)
+
+    def copy_udev_rule_files(self, source, location):
+        src_dir_path = source
+        for rules_file_name in listdir(src_dir_path):
+            rules_file_path = _resource_path(path.join(src_dir_path, rules_file_name))
+            copy(rules_file_path, location)
+    
+def _resource_path(relative_path):
+    if hasattr(sys, '_MEIPASS'):
+        return path.join(sys._MEIPASS, relative_path)
+    return path.join(relative_path)

test/run_tests.py

@@ -14,6 +14,7 @@ from test_trezor import trezor_test_suite
 from test_ledger import ledger_test_suite
 from test_digitalbitbox import digitalbitbox_test_suite
 from test_keepkey import keepkey_test_suite
+from test_udevrules import TestUdevRulesInstaller
 
 parser = argparse.ArgumentParser(description='Setup the testing environment and run automated tests')
 trezor_group = parser.add_mutually_exclusive_group()
@@ -40,6 +41,9 @@ suite = unittest.TestSuite()
 suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestDescriptor))
 suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestSegwitAddress))
 suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestPSBT))
+if sys.platform.startswith("linux"):
+    suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestUdevRulesInstaller))
+
 
 if not args.no_trezor or not args.no_coldcard or args.ledger or not args.no_bitbox or not args.no_keepkey:
     # Start bitcoind

test/test_udevrules.py

@@ -0,0 +1,39 @@
+#! /usr/bin/env python3
+
+import unittest
+import json
+import filecmp
+from os import makedirs, remove, removedirs, walk, path 
+from hwilib.cli import process_commands
+
+class TestUdevRulesInstaller(unittest.TestCase):
+    INSTALLATION_FOLDER = 'rules.d'
+    SOURCE_FOLDER = '../udev'
+
+    @classmethod
+    def setUpClass(cls):
+        # Create directory where copy the udev rules to.
+        makedirs(cls.INSTALLATION_FOLDER, exist_ok=True)
+
+    @classmethod
+    def tearDownClass(self):
+        for root, dirs, files in walk(self.INSTALLATION_FOLDER, topdown=False):
+            for name in files:
+                remove(path.join(root, name))
+        removedirs(self.INSTALLATION_FOLDER)
+
+    def test_rules_file_are_copied(self):
+        result = process_commands( ['installudevrules', '--source', self.SOURCE_FOLDER, '--location', self.INSTALLATION_FOLDER])
+        self.assertIn('error', result)
+        self.assertIn('code', result)
+        self.assertEqual(result['error'], 'Need to be root.')
+        self.assertEqual(result['code'], -16)
+        # Assert files wre copied
+        for root, dirs, files in walk(self.INSTALLATION_FOLDER, topdown=False):
+            for file_name in files:
+                src = path.join(self.SOURCE_FOLDER, file_name)
+                tgt = path.join(self.INSTALLATION_FOLDER, file_name)
+                self.assertTrue(filecmp.cmp(src, tgt))
+
+if __name__ == "__main__":
+    unittest.main()