Page MenuHomeVyOS Platform
Files F118305117

file.py
No OneTemporary
Actions

Size
14 KB
Referenced Files
None
Subscribers
None

file.py
View Options

#!/usr/bin/python3
# Copyright 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)

File Metadata

Mime Type
text/x-script.python
Expires
Fri, Oct 16, 11:59 PM (2 h, 11 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4408940
Default Alt Text
file.py (14 KB)

Event Timeline