From 08a3280b062af83ee50fa139d7827d954907886e Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Mon, 13 May 2024 21:52:13 +0200 Subject: Implement re-encryption support --- src/etoolkit/__init__.py | 2 +- src/etoolkit/__main__.py | 379 ++++++++++++++++++++++++++++++----------------- src/etoolkit/etoolkit.py | 110 +++++++++++--- 3 files changed, 340 insertions(+), 151 deletions(-) (limited to 'src') diff --git a/src/etoolkit/__init__.py b/src/etoolkit/__init__.py index 0ef5957..711c6ae 100644 --- a/src/etoolkit/__init__.py +++ b/src/etoolkit/__init__.py @@ -18,7 +18,7 @@ from .etoolkit import EtoolkitInstance, EtoolkitInstanceError __author__ = 'Simeon Simeonov' -__version__ = '1.3.0' +__version__ = '2.0.0' __license__ = 'GPL3' diff --git a/src/etoolkit/__main__.py b/src/etoolkit/__main__.py index 8b5bd55..a2f2f20 100644 --- a/src/etoolkit/__main__.py +++ b/src/etoolkit/__main__.py @@ -42,117 +42,244 @@ DEFAULT_LOG_LEVEL = logging.WARNING logger = logging.getLogger(__name__) -def decrypt_value(args: argparse.Namespace, config: dict): +class EtoolkitCLIHandler: """ - Interactive function for decrypting value(s) + Helper class used for handleing the growing amount of arguments - Prompts for master key password and then prompts for a value to decrypt + This class consists mostly of interactive methods and is not intended as + a part of the etoolkit API + """ - The decrypted value is printed to stdout + def __init__(self, args: argparse.Namespace, config_dict: dict): + """ + :param args: The parsed argparse arguments sent by the caller + :type args: argparse.Namespace + + :param config_dict: The config file structure + :type config_dict: dict + """ + self._args = args + self._config_dict = config_dict + + self._password_hash = None + if 'general' in config_dict: + self._password_hash = config_dict['general'].get( + 'MASTER_PASSWORD_HASH' + ) - :param args: The arguments sent by the caller - :type args: arparse.Namespace + self._password_from_env = os.environ.get('ETOOLKIT_MASTER_PASSWORD') - :param config: The config dict sent by the caller - :type config: dict - """ - password_hash = None - pipe_input = None - if not os.isatty(sys.stdin.fileno()): - pipe_input = sys.stdin.read().strip() - if 'general' in config: - password_hash = config['general'].get('MASTER_PASSWORD_HASH') - - if ( - args.master_password_prompt - or os.environ.get('ETOOLKIT_MASTER_PASSWORD') is None - ): - password = etoolkit.EtoolkitInstance.confirm_password_prompt( - password_hash, False - ) - else: - password = os.environ.get('ETOOLKIT_MASTER_PASSWORD') - - if pipe_input: - # the input came from stdin. No need to prompt - print( - 'Decrypted value: ' - f'{etoolkit.EtoolkitInstance.decrypt(password, pipe_input)}' - ) - return - while True: - try: - value = input('Value: ') + def decrypt_value(self): + """ + Interactive method for decrypting value(s) + + Prompts for master key password and then prompts for a value to decrypt + + The decrypted value is printed to stdout + """ + pipe_input = None + if not os.isatty(sys.stdin.fileno()): + pipe_input = sys.stdin.read().strip() + + if ( + self._args.master_password_prompt + or self._password_from_env is None + ): + password = self._password_prompt() + else: + password = self._password_from_env + + if pipe_input: + # the input came from stdin. No need to prompt print( 'Decrypted value: ' - f'{etoolkit.EtoolkitInstance.decrypt(password, value)}' + f'{etoolkit.EtoolkitInstance.decrypt(password, pipe_input)}' ) - if not args.multiple_values: + return + while True: + try: + value = input('Value: ') + print( + 'Decrypted value: ' + f'{etoolkit.EtoolkitInstance.decrypt(password, value)}' + ) + if not self._args.multiple_values: + break + except KeyboardInterrupt: + print(os.linesep) break - except KeyboardInterrupt: - print(os.linesep) - break - return + return + def encrypt_value(self): + """ + Interactive method for encrypting value(s) -def encrypt_value(args: argparse.Namespace, config: dict): - """ - Interactive function for encrypting value(s) + Prompts for master key password and then prompts for a value to encrypt + + The encrypted value is printed to stdout + """ + pipe_input = None + if not os.isatty(sys.stdin.fileno()): + pipe_input = sys.stdin.read().strip() + + if ( + self._args.master_password_prompt + or self._password_from_env is None + ): + password = self._password_prompt_confirm() + else: + password = self._password_from_env - Prompts for master key password and then prompts for a value to encrypt + if pipe_input: + # the input came from stdin. No need to prompt + print( + 'Encrypted value: ' + f'{etoolkit.EtoolkitInstance.encrypt(password, pipe_input)}' + ) + return + + while True: + try: + if self._args.echo: + value = input('Value: ') + else: + value = getpass.getpass('Value: ') + print( + 'Encrypted value: ' + f'{etoolkit.EtoolkitInstance.encrypt(password, value)}' + ) + if not self._args.multiple_values: + break + except KeyboardInterrupt: + print(os.linesep) + break + return - The encrypted value is printed to stdout + def generate_master_password_hash(self): + """ + Interactive method for generating password hash - :param args: The arguments sent by the caller - :type args: arparse.Namespace + Prompts for master key password and then for confirmation - :param config: The config dict sent by the caller - :type config: dict - """ - password_hash = None - pipe_input = None - if not os.isatty(sys.stdin.fileno()): - pipe_input = sys.stdin.read().strip() - if 'general' in config: - password_hash = config['general'].get('MASTER_PASSWORD_HASH') - - if ( - args.master_password_prompt - or os.environ.get('ETOOLKIT_MASTER_PASSWORD') is None - ): - password = etoolkit.EtoolkitInstance.confirm_password_prompt( - password_hash + The generated hash is printed to stdout + """ + phash = etoolkit.EtoolkitInstance.get_new_password_hash( + etoolkit.EtoolkitInstance.confirm_password_prompt() ) - else: - password = os.environ.get('ETOOLKIT_MASTER_PASSWORD') - - if pipe_input: - # the input came from stdin. No need to prompt - print( - 'Encrypted value: ' - f'{etoolkit.EtoolkitInstance.encrypt(password, pipe_input)}' + print(f'Master password hash: {phash}') + + def list(self): + """Lists all instances defined in the config file""" + + for instance_name in sorted( + filter( + lambda s: not s.startswith('_'), + self._config_dict.get('instances', {}).keys(), + ) + ): + print(instance_name) + + def load_instance(self): + """Loads a single specified instance from the config file""" + + inst = etoolkit.EtoolkitInstance( + self._args.instance, self._config_dict ) - return - while True: - try: - if args.echo: - value = input('Value: ') - else: - value = getpass.getpass('Value: ') + + if ( + self._args.master_password_prompt + or self._password_from_env is None + ): + inst.prompt_func = ( + etoolkit.EtoolkitInstance.confirm_password_prompt + ) + + env = inst.get_environ() + + if self._args.dump_output: + print(inst.env_to_str(env)) + + os.environ.update(env) + + if self._args.spawn: + subprocess.run(self._args.spawn.split(), check=False) + else: + subprocess.run( + os.environ.get('SHELL', 'bash').split(), check=False + ) + + def reencrypt(self): + """ + Interactive method that prints new configuration data (JSON) to stdout + + Prompts for master key password and then for a new password, + which may be the same as the current password + + All existing encrypted values are decrypted using the current password + and then encrypted with the new password + """ + print('(Current password) ', end='', flush=True) + if ( + self._args.master_password_prompt + or self._password_from_env is None + ): + password = self._password_prompt() + else: + password = self._password_from_env + + print('(New password) ', end='', flush=True) + new_password = etoolkit.EtoolkitInstance.confirm_password_prompt() + + if self._args.reencrypt != 'all': + # re-encrypt a single instance + inst = etoolkit.EtoolkitInstance( + self._args.reencrypt, self._config_dict + ) print( - 'Encrypted value: ' - f'{etoolkit.EtoolkitInstance.encrypt(password, value)}' + json.dumps( + inst.get_reencrypted_instance_data(new_password, password), + indent=4, + ) + ) + return + + # re-encrypt all + new_config_dict = dict(self._config_dict) + if ( + 'general' in new_config_dict + and 'MASTER_PASSWORD_HASH' in new_config_dict['general'] + ): + new_config_dict['general']['MASTER_PASSWORD_HASH'] = ( + etoolkit.EtoolkitInstance.get_new_password_hash(new_password) ) - if not args.multiple_values: - break - except KeyboardInterrupt: - print(os.linesep) - break - return + + for instance_name in self._config_dict['instances']: + inst = etoolkit.EtoolkitInstance(instance_name, self._config_dict) + new_config_dict['instances'][instance_name] = ( + inst.get_reencrypted_instance_data(new_password, password) + ) + print(json.dumps(new_config_dict, indent=4)) + + def _password_prompt(self) -> str: + """ + Wrapper for EtoolkitInstance.confirm_password_prompt(confirm=False) + """ + return etoolkit.EtoolkitInstance.confirm_password_prompt( + self._password_hash, False + ) + + def _password_prompt_confirm(self) -> str: + """ + Wrapper for EtoolkitInstance.confirm_password_prompt(confirm=True) + """ + return etoolkit.EtoolkitInstance.confirm_password_prompt( + self._password_hash + ) def main(inargs=None): """main entry point""" + parser = argparse.ArgumentParser( prog=__package__, epilog=( @@ -207,6 +334,20 @@ def main(inargs=None): required=False, help='Prompt for master password, display the generated hash and exit', ) + group.add_argument( + '-r', + '--reencrypt', + metavar='', + type=str, + default='', + dest='reencrypt', + required=False, + help=( + 'Prompt for current master password, new master password and ' + 're-encrypt either all encrypted values or only those for a ' + 'given instance' + ), + ) parser.add_argument( '-c', '--config-file', @@ -274,7 +415,7 @@ def main(inargs=None): try: with io.open(args.config_file, encoding='utf-8') as fp: config_dict = json.load(fp) - except FileNotFoundError as e: + except FileNotFoundError as err: # do not raise exception if config-file is missing for: # - decrypting value # - encrypting value @@ -288,66 +429,38 @@ def main(inargs=None): config_dict = {} else: logger.error('Configuration file %s is missing', args.config_file) - raise SystemExit(errno.EIO) from e - except Exception as e: + raise SystemExit(errno.EIO) from err + except Exception as exp: logger.exception('Unable to parse %r', args.config_file) - raise SystemExit(errno.EIO) from e + raise SystemExit(errno.EIO) from exp try: + etoolkit_cli_handler = EtoolkitCLIHandler(args, config_dict) if args.decrypt_value: - decrypt_value(args, config_dict) + etoolkit_cli_handler.decrypt_value() sys.exit(0) if args.encrypt_value: - encrypt_value(args, config_dict) + etoolkit_cli_handler.encrypt_value() sys.exit(0) if args.password_hash: - master_password = ( - etoolkit.EtoolkitInstance.confirm_password_prompt() - ) - phash = etoolkit.EtoolkitInstance.get_new_password_hash( - master_password - ) - print(f'Master password hash: {phash}') + etoolkit_cli_handler.generate_master_password_hash() sys.exit(0) if args.list: - for instance_name in sorted( - filter( - lambda s: not s.startswith('_'), - config_dict.get('instances', {}).keys(), - ) - ): - print(instance_name) + etoolkit_cli_handler.list() + sys.exit(0) + if args.reencrypt: + etoolkit_cli_handler.reencrypt() sys.exit(0) - inst = etoolkit.EtoolkitInstance(args.instance, config_dict) - if ( - args.master_password_prompt - or os.environ.get('ETOOLKIT_MASTER_PASSWORD') is None - ): - inst.prompt_func = ( - etoolkit.EtoolkitInstance.confirm_password_prompt - ) - env = inst.get_environ() - - if args.dump_output: - inst.dump_env(env) - - os.environ.update(env) - - if args.spawn: - subprocess.run(args.spawn.split(), check=False) - else: - subprocess.run( - os.environ.get('SHELL', 'bash').split(), check=False - ) + etoolkit_cli_handler.load_instance() except KeyboardInterrupt: logger.debug('KeyboardInterrupt') print(os.linesep) sys.exit(0) - except etoolkit.EtoolkitInstanceError as e: - logger.error('EtoolkitInstanceError: %s', e) + except etoolkit.EtoolkitInstanceError as err: + logger.error('EtoolkitInstanceError: %s', err) sys.exit(1) - except subprocess.CalledProcessError as e: - logger.error('Unable to spawn shell process: %s', e) + except subprocess.CalledProcessError as err: + logger.error('Unable to spawn shell process: %s', err) sys.exit(1) except Exception: logger.exception('Unexpected exception') diff --git a/src/etoolkit/etoolkit.py b/src/etoolkit/etoolkit.py index a2e0d7a..9a1b47e 100644 --- a/src/etoolkit/etoolkit.py +++ b/src/etoolkit/etoolkit.py @@ -23,7 +23,6 @@ import os from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.ciphers.aead import AESGCM - MIN_ENCRYPTED_VALUE_LENGTH = 32 @@ -50,24 +49,26 @@ class EtoolkitInstance: self._master_password_hash = None self._prompt_func = None # function to use when prompting for input try: - inst_data = data['instances'][name] - except KeyError as e: - raise EtoolkitInstanceError(f'Unknown instance "{name}"') from e - if inst_data.get('ETOOLKIT_PARENT'): - self._parent = EtoolkitInstance(inst_data['ETOOLKIT_PARENT'], data) + self._instance_data = data['instances'][name] + except KeyError as err: + raise EtoolkitInstanceError(f'Unknown instance "{name}"') from err + if self._instance_data.get('ETOOLKIT_PARENT'): + self._parent = EtoolkitInstance( + self._instance_data['ETOOLKIT_PARENT'], data + ) self._raw_env_variables.update(self._parent.raw_env_variables) self._sensitive_env_variables.extend( self._parent.sensitive_env_variables ) - if inst_data.get('ETOOLKIT_SENSITIVE'): - if not isinstance(inst_data['ETOOLKIT_SENSITIVE'], list): + if self._instance_data.get('ETOOLKIT_SENSITIVE'): + if not isinstance(self._instance_data['ETOOLKIT_SENSITIVE'], list): raise EtoolkitInstanceError( '"ETOOLKIT_SENSITIVE" must be a list' ) self._sensitive_env_variables.extend( - inst_data['ETOOLKIT_SENSITIVE'] + self._instance_data['ETOOLKIT_SENSITIVE'] ) - self._raw_env_variables.update(inst_data) + self._raw_env_variables.update(self._instance_data) # remove non env. variable data self._raw_env_variables.pop('ETOOLKIT_PARENT', None) self._raw_env_variables.pop('ETOOLKIT_SENSITIVE', None) @@ -200,7 +201,7 @@ class EtoolkitInstance: # padding_length_bytes(2 bytes) data padding (between 0 and 32) # extract padding_length_bytes - if data[:2] == b'--': + if data[:2] == b'-1' or data[:2] == b'--': data = data[2:] else: data = data[2 : -int(data[:2].decode())] @@ -259,7 +260,7 @@ class EtoolkitInstance: ) ) nonce = salt[:12] - padding_length_bytes = b'--' # no padding used 2 bytes "sign" + padding_length_bytes = b'-1' # no padding used 2 bytes "sign" edata = aesgcm.encrypt( nonce, padding_length_bytes + data_bytes, salt ) @@ -347,18 +348,49 @@ class EtoolkitInstance: except Exception: return False - def dump_env(self, env: dict): + @staticmethod + def reencrypt(password: str, new_password: str, edata: str) -> str: + """ + Re-encrypts `edata` using `password` and `new_password`. + + Version 2 of the etoolkit encryption format + + `edata` is in the following format: + enc-val$`version-num`$`bas64-salt`$`base64-encrypted_data` + + :param password: The password to decrypt `edata` with + :type password: str + + :param new_password: The password to re-encrypt the plain-text with + :type new_password: str + + :param edata: The data to be re-encrypted + :type edata: str + + :return: The new encrypted string string + :rtype: str + """ + return EtoolkitInstance.encrypt( + new_password, EtoolkitInstance.decrypt(password, edata) + ) + + def env_to_str(self, env: dict) -> str: """ - Prints an environment dict to stdout. + Returns a printable str. representation of the environment dict :param env: The environment dict :type env: dict + + :return: Printable representation of the environment dict + :rtype: str """ + env_str = '' for key, value in env.items(): if key in self._sensitive_env_variables: - print(f'{key}: ***') + env_str += f'{key}: ***{os.linesep}' continue - print(f'{key}: {value}') + env_str += f'{key}: {value}{os.linesep}' + return env_str def get_environ(self) -> dict: """ @@ -421,6 +453,50 @@ class EtoolkitInstance: return self.name return self._parent.get_full_name(delimiter) + delimiter + self.name + def get_reencrypted_instance_data( + self, new_password: str, password: str = None + ) -> dict: + """ + Returns new instance data (dict) containing new encrypted values + + Each encrypted value in this instance is decrypted using `password` + and then encrypted again using `new_password` + + If `password` is None, master_password is not set earlier for this + instance and 'ETOOLKIT_MASTER_PASSWORD' is not set, + the prompt function will be called + + :param new_password: The password to reencrypt with + :type new_password: str + + :param password: The password to decrypt current encrypted values with + :type password: str or None + + :return: New instance data + :rtype: dict + """ + if password is None: + password = self._master_password + + if password is None and self._prompt_func is None: + password = os.environ.get('ETOOLKIT_MASTER_PASSWORD') + if password is None: + raise EtoolkitInstanceError( + 'Neither password or prompt function set' + ) + + if password is None: + password = self._prompt_func( + self._master_password_hash, confirm=False + ) + + new_data = dict(self._instance_data) + for key, value in self._instance_data.items(): + if isinstance(value, str) and value.startswith('enc-val$'): + new_data[key] = self.reencrypt(password, new_password, value) + + return new_data + def _decrypt_value(self, evalue: str) -> str: """ Decrypts an encrypted value using the master password @@ -452,4 +528,4 @@ class EtoolkitInstance: self.master_password = self._prompt_func( self._master_password_hash, confirm=False ) - return EtoolkitInstance.decrypt(self._master_password, evalue) + return self.decrypt(self._master_password, evalue) -- cgit v1.3