diff options
Diffstat (limited to 'src')
| -rwxr-xr-x | src/completion/list_images.py | 44 | ||||
| -rwxr-xr-x | src/op_mode/file.py | 383 |
2 files changed, 427 insertions, 0 deletions
diff --git a/src/completion/list_images.py b/src/completion/list_images.py new file mode 100755 index 000000000..eae29c084 --- /dev/null +++ b/src/completion/list_images.py @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2023 VyOS maintainers and contributors +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 or later as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +import argparse +import os +import sys + +from vyos.system.image import is_live_boot +from vyos.system.image import get_running_image + + +parser = argparse.ArgumentParser(description='list available system images') +parser.add_argument('--no-running', action='store_true', + help='do not display the currently running image') + +def get_images(omit_running: bool = False) -> list[str]: + if is_live_boot(): + return [] + images = os.listdir("/lib/live/mount/persistence/boot") + if omit_running: + images.remove(get_running_image()) + if 'grub' in images: + images.remove('grub') + if 'efi' in images: + images.remove('efi') + return sorted(images) + +if __name__ == '__main__': + args = parser.parse_args() + print("\n".join(get_images(omit_running=args.no_running))) + sys.exit(0) diff --git a/src/op_mode/file.py b/src/op_mode/file.py new file mode 100755 index 000000000..bf13bed6f --- /dev/null +++ b/src/op_mode/file.py @@ -0,0 +1,383 @@ +#!/usr/bin/python3 + +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import argparse +import contextlib +import datetime +import grp +import os +import pwd +import shutil +import sys +import tempfile + +from vyos.remote import download +from vyos.remote import upload +from vyos.utils.io import ask_yes_no +from vyos.utils.io import print_error +from vyos.utils.process import cmd +from vyos.utils.process import run + + +parser = argparse.ArgumentParser(description='view, copy or remove files and directories', + formatter_class=argparse.RawDescriptionHelpFormatter) +parser.epilog = """ +TYPE is one of 'remote', 'image' and 'local'. +A local path is <path> or ~/<path>. +A remote path is <scheme>://<urn>. +An image path is <image>:<path>. + +Clone operation is between images only. +Copy operation does not support directories from remote locations. +Delete operation does not support remote paths. +""" +operations = parser.add_mutually_exclusive_group(required=True) +operations.add_argument('--show', nargs=1, help='show the contents of file PATH of type TYPE', + metavar=('PATH')) +operations.add_argument('--copy', nargs=2, help='copy SRC to DEST', + metavar=('SRC', 'DEST')) +operations.add_argument('--delete', nargs=1, help='delete file PATH', + metavar=('PATH')) +operations.add_argument('--clone', help='clone config from running image to IMG', + metavar='IMG') +operations.add_argument('--clone-from', nargs=2, help='clone config from image SRC to image DEST', + metavar=('SRC', 'DEST')) + +## Helper procedures +def fix_terminal() -> None: + """ + Reset terminal after potential breakage caused by abrupt exits. + """ + run('stty sane') + +def get_types(arg: str) -> tuple[str, str]: + """ + Determine whether the argument shows a local, image or remote path. + """ + schemes = ['http', 'https', 'ftp', 'ftps', 'sftp', 'ssh', 'scp', 'tftp'] + s = arg.split("://", 1) + if len(s) != 2: + return 'local', arg + elif s[0] in schemes: + return 'remote', arg + else: + return 'image', arg + +def zealous_copy(source: str, destination: str) -> None: + # Even shutil.copy2() doesn't preserve ownership across copies. + # So we need to resort to this. + stats = os.stat(source) + shutil.copy2(source, destination) + os.chown(destination, stats.st_uid, stats.st_gid) + +def get_file_type(path: str) -> str: + return cmd(['file', '-sb', path]) + +def print_header(string: str) -> None: + print('#' * 10, string, '#' * 10) + +def octal_to_symbolic(octal: str) -> str: + perms = ['---', '--x', '-w-', '-wx', 'r--', 'r-x', 'rw-', 'rwx'] + result = "" + # We discard all but the last three digits because we're only + # interested in the permission bits. + for i in octal[-3:]: + result += perms[int(i)] + return result + +def get_user_and_group(stats: os.stat_result) -> tuple[str, str]: + try: + user = pwd.getpwuid(stats.st_uid).pw_name + except (KeyError, PermissionError): + user = str(stats.st_uid) + try: + group = grp.getgrgid(stats.st_gid).gr_name + except (KeyError, PermissionError): + group = str(stats.st_gid) + return user, group + +def print_file_info(path: str) -> None: + stats = os.stat(path) + username, groupname = get_user_and_group(stats) + mtime = datetime.datetime.fromtimestamp(stats.st_mtime).strftime("%F %X") + print_header('FILE INFO') + print(f'Path:\t\t{path}') + # File type is determined through `file(1)`. + print(f'Type:\t\t{get_file_type(path)}') + # Owner user and group + print(f'Owner:\t\t{username}:{groupname}') + # Permissions are converted from raw int to octal string to symbolic string. + print(f'Permissions:\t{octal_to_symbolic(oct(stats.st_mode))}') + # Last date of modification + print(f'Modified:\t{mtime}') + +def print_file_data(path: str) -> None: + print_header('FILE DATA') + file_type = get_file_type(path) + # Human-readable files are streamed line-by-line. + if 'text' in file_type: + with open(path, 'r') as f: + for line in f: + print(line, end='') + # tcpdump files go to TShark. + elif 'pcap' in file_type or os.path.splitext(path)[1] == '.pcap': + print(cmd(['sudo', 'tshark', '-r', path])) + # All other binaries get hexdumped. + else: + print(cmd(['hexdump', '-C', path])) + +def parse_image_path(image_path: str) -> str: + """ + my-image:/foo/bar -> /lib/live/mount/persistence/boot/my-image/rw/foo/bar + """ + image_name, path = image_path.split('://', 1) + if image_name == 'running': + image_root = '/' + elif image_name == 'disk-install': + image_root = '/lib/live/mount/persistence/' + else: + image_root = os.path.join('/lib/live/mount/persistence/boot', image_name, 'rw') + if not os.path.isdir(image_root): + print_error(f'Image {image_name} not found.') + sys.exit(1) + return os.path.join(image_root, path) + + +## Show procedures +def show_locally(path: str) -> None: + """ + Display the contents of a local file or directory. + """ + location = os.path.realpath(os.path.expanduser(path)) + # Temporarily redirect stdout to a throwaway file for `less(1)` to read. + # The output could be potentially too hefty for an in-memory StringIO. + temp = tempfile.NamedTemporaryFile('w', delete=False) + try: + with contextlib.redirect_stdout(temp): + # Just a directory. Call `ls(1)` and bail. + if os.path.isdir(location): + print_header('DIRECTORY LISTING') + print('Path:\t', location) + print(cmd(['ls', '-hlFGL', '--group-directories-first', location])) + elif os.path.isfile(location): + print_file_info(location) + print() + print_file_data(location) + else: + print_error(f'File or directory {path} not found.') + sys.exit(1) + sys.stdout.flush() + # Call `less(1)` and wait for it to terminate before going forward. + cmd(['/usr/bin/less', '-X', temp.name], stdout=sys.stdout) + # The stream to the temporary file could break for any reason. + # It's much less fragile than if we streamed directly to the process stdin. + # But anything could still happen and we don't want to scare the user. + except (BrokenPipeError, EOFError, KeyboardInterrupt, OSError): + fix_terminal() + sys.exit(1) + finally: + os.remove(temp.name) + +def show(type: str, path: str) -> None: + if type == 'remote': + temp = tempfile.NamedTemporaryFile(delete=False) + download(temp.name, path) + show_locally(temp.name) + os.remove(temp.name) + elif type == 'image': + show_locally(parse_image_path(path)) + elif type == 'local': + show_locally(path) + else: + print_error(f'Unknown target for showing: {type}') + print_error('Valid types are "remote", "image" and "local".') + sys.exit(1) + + +## Copying procedures +def copy(source_type: str, source_path: str, + destination_type: str, destination_path: str) -> None: + """ + Copy a file or directory locally, remotely or to and from an image. + Directory uploads and downloads not supported. + """ + source = '' + try: + # Download to a temporary file and use that as the source. + if source_type == 'remote': + source = tempfile.NamedTemporaryFile(delete=False).name + download(source, source_path) + # Prepend the image root to the path. + elif source_type == 'image': + source = parse_image_path(source_path) + elif source_type == 'local': + source = source_path + else: + print_error(f'Unknown source type: {source_type}') + print_error(f'Valid source types are "remote", "image" and "local".') + sys.exit(1) + + # Directly upload the file. + if destination_type == 'remote': + if os.path.isdir(source): + print_error(f'Cannot upload {source}. Directory uploads not supported.') + sys.exit(1) + upload(source, destination_path) + # No need to duplicate local copy operations for image copying. + elif destination_type == 'image': + copy('local', source, 'local', parse_image_path(destination_path)) + # Try to preserve metadata when copying. + elif destination_type == 'local': + if os.path.isdir(destination_path): + destination_path = os.path.join(destination_path, os.path.basename(source)) + if os.path.isdir(source): + shutil.copytree(source, destination_path, copy_function=zealous_copy) + else: + zealous_copy(source, destination_path) + else: + print_error(f'Unknown destination type: {source_type}') + print_error(f'Valid destination types are "remote", "image" and "local".') + sys.exit(1) + except OSError: + import traceback + # We can't check for every single user error (eg copying a directory to a file) + # so we just let a curtailed stack trace provide a descriptive error. + print_error(f'Failed to copy {source_path} to {destination_path}.') + traceback.print_exception(*sys.exc_info()[:2], None) + sys.exit(1) + else: + # To prevent a duplicate message. + if destination_type != 'image': + print('Copy successful.') + finally: + # Clean up temporary file. + if source_type == 'remote': + os.remove(source) + + +## Deletion procedures +def delete_locally(path: str) -> None: + """ + Remove a local file or directory. + """ + try: + if os.path.isdir(path): + if (ask_yes_no(f'Do you want to remove {path} with all its contents?')): + shutil.rmtree(path) + print(f'Directory {path} removed.') + else: + print('Operation aborted.') + elif os.path.isfile(path): + if (ask_yes_no(f'Do you want to remove {path}?')): + os.remove(path) + print(f'File {path} removed.') + else: + print('Operation aborted.') + else: + raise OSError(f'File or directory {path} not found.') + except OSError: + import traceback + print_error(f'Failed to delete {path}.') + traceback.print_exception(*sys.exc_info()[:2], None) + sys.exit(1) + +def delete(type: str, path: str) -> None: + if type == 'local': + delete_locally(path) + elif type == 'image': + delete_locally(parse_image_path(path)) + else: + print_error(f'Unknown target for deletion: {type}') + print_error('Valid types are "image" and "local".') + sys.exit(1) + + +## Cloning procedures +def clone(source: str, destination: str) -> None: + if os.geteuid(): + print_error('Only the superuser can run this command.') + sys.exit(1) + if destination == 'running' or destination == 'disk-install': + print_error(f'Cannot clone config to {destination}.') + sys.exit(1) + # If `source` is None, then we're going to copy from the running image. + if source is None or source == 'running': + source_path = '/config' + # For the warning message only. + source = 'the current' + else: + source_path = parse_image_path(source + ':/config') + destination_path = parse_image_path(destination + ':/config') + backup_path = destination_path + '.preclone' + + if not os.path.isdir(source_path): + print_error(f'Source image {source} does not exist.') + sys.exit(1) + if not os.path.isdir(destination_path): + print_error(f'Destination image {destination} does not exist.') + sys.exit(1) + print(f'WARNING: This operation will erase /config data in image {destination}.') + print(f'/config data in {source} image will be copied over in its place.') + print(f'The existing /config data in {destination} image will be backed up to /config.preclone.') + + if ask_yes_no('Are you sure you want to continue?'): + try: + if os.path.isdir(backup_path): + print('Removing previous backup...') + shutil.rmtree(backup_path) + print('Making new backup...') + shutil.move(destination_path, backup_path) + except: + print('Something went wrong during the backup process!') + print('Cowardly refusing to proceed with cloning.') + raise + # Copy new config from image. + try: + shutil.copytree(source_path, destination_path, copy_function=zealous_copy) + except: + print('Cloning failed! Reverting to backup!') + # Delete leftover files from the botched cloning. + shutil.rmtree(destination_path, ignore_errors=True) + # Restore backup before bailing out. + shutil.copytree(backup_path, destination_path, copy_function=zealous_copy) + raise + else: + print(f'Successfully cloned config from {source} to {destination}.') + finally: + shutil.rmtree(backup_path) + else: + print('Operation aborted.') + +if __name__ == '__main__': + args = parser.parse_args() + try: + if args.show: + show(*get_types(args.show[0])) + elif args.copy: + copy(*get_types(args.copy[0]), + *get_types(args.copy[1])) + elif args.delete: + delete(*get_types(args.delete[0])) + elif args.clone_from: + clone(*args.clone_from) + elif args.clone: + # Pass None as source image to copy from local image. + clone(None, args.clone) + except KeyboardInterrupt: + print_error('Operation cancelled by user.') + sys.exit(1) + sys.exit(0) |