Loading domain_substitution/_common.py +1 −2 Original line number Diff line number Diff line Loading @@ -36,9 +36,8 @@ class ExtractorEnum: #pylint: disable=too-few-public-methods class SetLogLevel(argparse.Action): #pylint: disable=too-few-public-methods """Sets logging level based on command line arguments it receives""" def __init__(self, option_strings, dest, nargs=None, **kwargs): super(SetLogLevel, self).__init__(option_strings, dest, nargs=nargs, **kwargs) super().__init__(option_strings, dest, nargs=nargs, **kwargs) def __call__(self, parser, namespace, value, option_string=None): if option_string in ('--verbose', '-v'): Loading domain_substitution/_extraction.py +22 −37 Original line number Diff line number Diff line Loading @@ -23,24 +23,18 @@ DEFAULT_EXTRACTORS = { } class ExtractionError(BaseException): """Exceptions thrown in this module's methods""" def _find_7z_by_registry(): """ Return a string to 7-zip's 7z.exe from the Windows Registry. Raises ExtractionError if it fails. """ import winreg #pylint: disable=import-error import winreg #pylint: disable=import-error, import-outside-toplevel sub_key_7zfm = 'SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\App Paths\\7zFM.exe' try: with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_key_7zfm) as key_handle: sevenzipfm_dir = winreg.QueryValueEx(key_handle, 'Path')[0] except OSError: get_logger().exception('Unable to locate 7-zip from the Windows Registry') raise ExtractionError() raise sevenzip_path = Path(sevenzipfm_dir, '7z.exe') if not sevenzip_path.is_file(): get_logger().error('7z.exe not found at path from registry: %s', sevenzip_path) Loading @@ -50,17 +44,15 @@ def _find_7z_by_registry(): def _find_winrar_by_registry(): """ Return a string to WinRAR's WinRAR.exe from the Windows Registry. Raises ExtractionError if it fails. """ import winreg #pylint: disable=import-error import winreg #pylint: disable=import-error, import-outside-toplevel sub_key_winrar = 'SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\App Paths\\WinRAR.exe' try: with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_key_winrar) as key_handle: winrar_dir = winreg.QueryValueEx(key_handle, 'Path')[0] except OSError: get_logger().exception('Unable to locale WinRAR from the Windows Registry') raise ExtractionError() raise winrar_path = Path(winrar_dir, 'WinRAR.exe') if not winrar_path.is_file(): get_logger().error('WinRAR.exe not found at path from registry: %s', winrar_path) Loading Loading @@ -89,7 +81,7 @@ def _process_relative_to(unpack_root, relative_to): if not relative_root.is_dir(): get_logger().error('Could not find relative_to directory in extracted files: %s', relative_to) raise ExtractionError() raise Exception() for src_path in relative_root.iterdir(): dest_path = unpack_root / src_path.name src_path.rename(dest_path) Loading @@ -101,7 +93,7 @@ def _extract_tar_with_7z(binary, archive_path, output_dir, relative_to, skip_unu if not relative_to is None and (output_dir / relative_to).exists(): get_logger().error('Temporary unpacking directory already exists: %s', output_dir / relative_to) raise ExtractionError() raise Exception() cmd1 = (binary, 'x', str(archive_path), '-so') cmd2 = (binary, 'x', '-si', '-aoa', '-ttar', '-o{}'.format(str(output_dir))) if skip_unused: Loading @@ -117,7 +109,7 @@ def _extract_tar_with_7z(binary, archive_path, output_dir, relative_to, skip_unu get_logger().error('7z commands returned non-zero status: %s', proc2.returncode) get_logger().debug('stdout: %s', stdout_data) get_logger().debug('stderr: %s', stderr_data) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading @@ -130,10 +122,10 @@ def _extract_tar_with_tar(binary, archive_path, output_dir, relative_to, skip_un for cpath in CONTINGENT_PATHS: cmd += ('--exclude=%s/%s' % (str(relative_to), cpath[:-1]), ) get_logger().debug('tar command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('tar command returned %s', result.returncode) raise ExtractionError() raise Exception() # for gnu tar, the --transform option could be used. but to keep compatibility with # bsdtar on macos, we just do this ourselves Loading @@ -146,12 +138,12 @@ def _extract_tar_with_winrar(binary, archive_path, output_dir, relative_to, skip cmd = (binary, 'x', '-o+', str(archive_path), str(output_dir)) if skip_unused: for cpath in CONTINGENT_PATHS: cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/'), os.sep), ) cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/')), ) get_logger().debug('WinRAR command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('WinRAR command returned %s', result.returncode) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading @@ -161,7 +153,6 @@ def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused) class NoAppendList(list): """Hack to workaround memory issues with large tar files""" def append(self, obj): pass Loading @@ -178,7 +169,7 @@ def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused) except BaseException: # Unexpected exception get_logger().exception('Unexpected exception during symlink support check.') raise ExtractionError() raise with tarfile.open(str(archive_path), 'r|%s' % archive_path.suffix[1:]) as tar_file_obj: tar_file_obj.members = NoAppendList() Loading Loading @@ -209,7 +200,7 @@ def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused) tar_file_obj._extract_member(tarinfo, str(destination)) # pylint: disable=protected-access except BaseException: get_logger().exception('Exception thrown for tar member: %s', tarinfo.name) raise ExtractionError() raise def extract_tar_file(archive_path, output_dir, relative_to, skip_unused, extractors=None): Loading @@ -223,8 +214,6 @@ def extract_tar_file(archive_path, output_dir, relative_to, skip_unused, extract root of the archive, or None if no path components should be stripped. extractors is a dictionary of PlatformEnum to a command or path to the extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip and WinRAR. Raises ExtractionError if unexpected issues arise during unpacking. """ if extractors is None: extractors = DEFAULT_EXTRACTORS Loading Loading @@ -280,8 +269,6 @@ def extract_with_7z( root of the archive. extractors is a dictionary of PlatformEnum to a command or path to the extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip. Raises ExtractionError if unexpected issues arise during unpacking. """ # TODO: It would be nice to extend this to support arbitrary standard IO chaining of 7z # instances, so _extract_tar_with_7z and other future formats could use this. Loading @@ -291,24 +278,24 @@ def extract_with_7z( if sevenzip_cmd == USE_REGISTRY: if not get_running_platform() == PlatformEnum.WINDOWS: get_logger().error('"%s" for 7-zip is only available on Windows', sevenzip_cmd) raise ExtractionError() raise Exception() sevenzip_cmd = str(_find_7z_by_registry()) sevenzip_bin = _find_extractor_by_cmd(sevenzip_cmd) if not relative_to is None and (output_dir / relative_to).exists(): get_logger().error('Temporary unpacking directory already exists: %s', output_dir / relative_to) raise ExtractionError() raise Exception() cmd = (sevenzip_bin, 'x', str(archive_path), '-aoa', '-o{}'.format(str(output_dir))) if skip_unused: for cpath in CONTINGENT_PATHS: cmd += ('-x!%s/%s' % (str(relative_to), cpath[:-1]), ) get_logger().debug('7z command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('7z command returned %s', result.returncode) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading @@ -330,8 +317,6 @@ def extract_with_winrar( root of the archive. extractors is a dictionary of PlatformEnum to a command or path to the extractor binary. Defaults to 'tar' for tar, and '_use_registry' for WinRAR. Raises ExtractionError if unexpected issues arise during unpacking. """ if extractors is None: extractors = DEFAULT_EXTRACTORS Loading @@ -339,23 +324,23 @@ def extract_with_winrar( if winrar_cmd == USE_REGISTRY: if not get_running_platform() == PlatformEnum.WINDOWS: get_logger().error('"%s" for WinRAR is only available on Windows', winrar_cmd) raise ExtractionError() raise Exception() winrar_cmd = str(_find_winrar_by_registry()) winrar_bin = _find_extractor_by_cmd(winrar_cmd) if not relative_to is None and (output_dir / relative_to).exists(): get_logger().error('Temporary unpacking directory already exists: %s', output_dir / relative_to) raise ExtractionError() raise Exception() cmd = (winrar_bin, 'x', '-o+', str(archive_path), str(output_dir)) if skip_unused: for cpath in CONTINGENT_PATHS: cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/', os.sep)), ) get_logger().debug('WinRAR command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('WinRAR command returned %s', result.returncode) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading
domain_substitution/_common.py +1 −2 Original line number Diff line number Diff line Loading @@ -36,9 +36,8 @@ class ExtractorEnum: #pylint: disable=too-few-public-methods class SetLogLevel(argparse.Action): #pylint: disable=too-few-public-methods """Sets logging level based on command line arguments it receives""" def __init__(self, option_strings, dest, nargs=None, **kwargs): super(SetLogLevel, self).__init__(option_strings, dest, nargs=nargs, **kwargs) super().__init__(option_strings, dest, nargs=nargs, **kwargs) def __call__(self, parser, namespace, value, option_string=None): if option_string in ('--verbose', '-v'): Loading
domain_substitution/_extraction.py +22 −37 Original line number Diff line number Diff line Loading @@ -23,24 +23,18 @@ DEFAULT_EXTRACTORS = { } class ExtractionError(BaseException): """Exceptions thrown in this module's methods""" def _find_7z_by_registry(): """ Return a string to 7-zip's 7z.exe from the Windows Registry. Raises ExtractionError if it fails. """ import winreg #pylint: disable=import-error import winreg #pylint: disable=import-error, import-outside-toplevel sub_key_7zfm = 'SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\App Paths\\7zFM.exe' try: with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_key_7zfm) as key_handle: sevenzipfm_dir = winreg.QueryValueEx(key_handle, 'Path')[0] except OSError: get_logger().exception('Unable to locate 7-zip from the Windows Registry') raise ExtractionError() raise sevenzip_path = Path(sevenzipfm_dir, '7z.exe') if not sevenzip_path.is_file(): get_logger().error('7z.exe not found at path from registry: %s', sevenzip_path) Loading @@ -50,17 +44,15 @@ def _find_7z_by_registry(): def _find_winrar_by_registry(): """ Return a string to WinRAR's WinRAR.exe from the Windows Registry. Raises ExtractionError if it fails. """ import winreg #pylint: disable=import-error import winreg #pylint: disable=import-error, import-outside-toplevel sub_key_winrar = 'SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\App Paths\\WinRAR.exe' try: with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_key_winrar) as key_handle: winrar_dir = winreg.QueryValueEx(key_handle, 'Path')[0] except OSError: get_logger().exception('Unable to locale WinRAR from the Windows Registry') raise ExtractionError() raise winrar_path = Path(winrar_dir, 'WinRAR.exe') if not winrar_path.is_file(): get_logger().error('WinRAR.exe not found at path from registry: %s', winrar_path) Loading Loading @@ -89,7 +81,7 @@ def _process_relative_to(unpack_root, relative_to): if not relative_root.is_dir(): get_logger().error('Could not find relative_to directory in extracted files: %s', relative_to) raise ExtractionError() raise Exception() for src_path in relative_root.iterdir(): dest_path = unpack_root / src_path.name src_path.rename(dest_path) Loading @@ -101,7 +93,7 @@ def _extract_tar_with_7z(binary, archive_path, output_dir, relative_to, skip_unu if not relative_to is None and (output_dir / relative_to).exists(): get_logger().error('Temporary unpacking directory already exists: %s', output_dir / relative_to) raise ExtractionError() raise Exception() cmd1 = (binary, 'x', str(archive_path), '-so') cmd2 = (binary, 'x', '-si', '-aoa', '-ttar', '-o{}'.format(str(output_dir))) if skip_unused: Loading @@ -117,7 +109,7 @@ def _extract_tar_with_7z(binary, archive_path, output_dir, relative_to, skip_unu get_logger().error('7z commands returned non-zero status: %s', proc2.returncode) get_logger().debug('stdout: %s', stdout_data) get_logger().debug('stderr: %s', stderr_data) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading @@ -130,10 +122,10 @@ def _extract_tar_with_tar(binary, archive_path, output_dir, relative_to, skip_un for cpath in CONTINGENT_PATHS: cmd += ('--exclude=%s/%s' % (str(relative_to), cpath[:-1]), ) get_logger().debug('tar command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('tar command returned %s', result.returncode) raise ExtractionError() raise Exception() # for gnu tar, the --transform option could be used. but to keep compatibility with # bsdtar on macos, we just do this ourselves Loading @@ -146,12 +138,12 @@ def _extract_tar_with_winrar(binary, archive_path, output_dir, relative_to, skip cmd = (binary, 'x', '-o+', str(archive_path), str(output_dir)) if skip_unused: for cpath in CONTINGENT_PATHS: cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/'), os.sep), ) cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/')), ) get_logger().debug('WinRAR command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('WinRAR command returned %s', result.returncode) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading @@ -161,7 +153,6 @@ def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused) class NoAppendList(list): """Hack to workaround memory issues with large tar files""" def append(self, obj): pass Loading @@ -178,7 +169,7 @@ def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused) except BaseException: # Unexpected exception get_logger().exception('Unexpected exception during symlink support check.') raise ExtractionError() raise with tarfile.open(str(archive_path), 'r|%s' % archive_path.suffix[1:]) as tar_file_obj: tar_file_obj.members = NoAppendList() Loading Loading @@ -209,7 +200,7 @@ def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused) tar_file_obj._extract_member(tarinfo, str(destination)) # pylint: disable=protected-access except BaseException: get_logger().exception('Exception thrown for tar member: %s', tarinfo.name) raise ExtractionError() raise def extract_tar_file(archive_path, output_dir, relative_to, skip_unused, extractors=None): Loading @@ -223,8 +214,6 @@ def extract_tar_file(archive_path, output_dir, relative_to, skip_unused, extract root of the archive, or None if no path components should be stripped. extractors is a dictionary of PlatformEnum to a command or path to the extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip and WinRAR. Raises ExtractionError if unexpected issues arise during unpacking. """ if extractors is None: extractors = DEFAULT_EXTRACTORS Loading Loading @@ -280,8 +269,6 @@ def extract_with_7z( root of the archive. extractors is a dictionary of PlatformEnum to a command or path to the extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip. Raises ExtractionError if unexpected issues arise during unpacking. """ # TODO: It would be nice to extend this to support arbitrary standard IO chaining of 7z # instances, so _extract_tar_with_7z and other future formats could use this. Loading @@ -291,24 +278,24 @@ def extract_with_7z( if sevenzip_cmd == USE_REGISTRY: if not get_running_platform() == PlatformEnum.WINDOWS: get_logger().error('"%s" for 7-zip is only available on Windows', sevenzip_cmd) raise ExtractionError() raise Exception() sevenzip_cmd = str(_find_7z_by_registry()) sevenzip_bin = _find_extractor_by_cmd(sevenzip_cmd) if not relative_to is None and (output_dir / relative_to).exists(): get_logger().error('Temporary unpacking directory already exists: %s', output_dir / relative_to) raise ExtractionError() raise Exception() cmd = (sevenzip_bin, 'x', str(archive_path), '-aoa', '-o{}'.format(str(output_dir))) if skip_unused: for cpath in CONTINGENT_PATHS: cmd += ('-x!%s/%s' % (str(relative_to), cpath[:-1]), ) get_logger().debug('7z command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('7z command returned %s', result.returncode) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to) Loading @@ -330,8 +317,6 @@ def extract_with_winrar( root of the archive. extractors is a dictionary of PlatformEnum to a command or path to the extractor binary. Defaults to 'tar' for tar, and '_use_registry' for WinRAR. Raises ExtractionError if unexpected issues arise during unpacking. """ if extractors is None: extractors = DEFAULT_EXTRACTORS Loading @@ -339,23 +324,23 @@ def extract_with_winrar( if winrar_cmd == USE_REGISTRY: if not get_running_platform() == PlatformEnum.WINDOWS: get_logger().error('"%s" for WinRAR is only available on Windows', winrar_cmd) raise ExtractionError() raise Exception() winrar_cmd = str(_find_winrar_by_registry()) winrar_bin = _find_extractor_by_cmd(winrar_cmd) if not relative_to is None and (output_dir / relative_to).exists(): get_logger().error('Temporary unpacking directory already exists: %s', output_dir / relative_to) raise ExtractionError() raise Exception() cmd = (winrar_bin, 'x', '-o+', str(archive_path), str(output_dir)) if skip_unused: for cpath in CONTINGENT_PATHS: cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/', os.sep)), ) get_logger().debug('WinRAR command line: %s', ' '.join(cmd)) result = subprocess.run(cmd) result = subprocess.run(cmd, check=False) if result.returncode != 0: get_logger().error('WinRAR command returned %s', result.returncode) raise ExtractionError() raise Exception() _process_relative_to(output_dir, relative_to)