Qubes/core-admin-client
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fix pylint warnings.

Browse Source
This commit is contained in:
WillyPillow 2020-07-11 23:12:57 +08:00
parent 8a4b5e683a
commit faef52e61a
2 changed files with 50 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
qubesadmin/tools/.qvm_template.py.swp
View File

Binary file not shown.

95
qubesadmin/tools/qvm_template.py
View File

@ -3,7 +3,6 @@
import argparse import argparse
import datetime import datetime
import enum import enum
import math
import os import os
import shutil import shutil
import subprocess import subprocess
@ -11,7 +10,6 @@ import sys
import tempfile import tempfile
import time import time
import dnf
import qubesadmin import qubesadmin
import qubesadmin.tools import qubesadmin.tools
import rpm import rpm
@ -26,14 +24,14 @@ UNVERIFIED_SUFFIX = '.unverified'
def qubes_release(): def qubes_release():
if os.path.exists('/usr/share/qubes/marker-vm'): if os.path.exists('/usr/share/qubes/marker-vm'):
with open('/usr/share/qubes/marker-vm', 'r') as f: with open('/usr/share/qubes/marker-vm', 'r') as fd:
# Get last line (in the format `x.x`) # Get last line (in the format `x.x`)
return f.readlines()[-1].strip() return fd.readlines()[-1].strip()
return subprocess.check_output(['lsb_release', '-sr'], return subprocess.check_output(['lsb_release', '-sr'],
encoding='UTF-8').strip() encoding='UTF-8').strip()
parser = argparse.ArgumentParser(description='Qubes Template Manager') parser = argparse.ArgumentParser(description='Qubes Template Manager')
parser.add_argument('operation', type=str) parser.add_argument('operation', type=str)
parser.add_argument('templates', nargs='*') parser.add_argument('templates', nargs='*')
# qrexec related # qrexec related
@ -82,9 +80,9 @@ class TemplateState(enum.Enum):
def verify_rpm(path, nogpgcheck=False, transaction_set=None): def verify_rpm(path, nogpgcheck=False, transaction_set=None):
if transaction_set is None: if transaction_set is None:
transaction_set = rpm.TransactionSet() transaction_set = rpm.TransactionSet()
with open(path, 'rb') as f: with open(path, 'rb') as fd:
try: try:
hdr = transaction_set.hdrFromFdno(f) hdr = transaction_set.hdrFromFdno(fd)
if hdr[rpm.RPMTAG_SIGSIZE] is None \ if hdr[rpm.RPMTAG_SIGSIZE] is None \
and hdr[rpm.RPMTAG_SIGPGP] is None \ and hdr[rpm.RPMTAG_SIGPGP] is None \
and hdr[rpm.RPMTAG_SIGGPG] is None: and hdr[rpm.RPMTAG_SIGGPG] is None:
@ -99,8 +97,8 @@ def verify_rpm(path, nogpgcheck=False, transaction_set=None):
def get_package_hdr(path, transaction_set=None): def get_package_hdr(path, transaction_set=None):
if transaction_set is None: if transaction_set is None:
transaction_set = rpm.TransactionSet() transaction_set = rpm.TransactionSet()
with open(path, 'rb') as f: with open(path, 'rb') as fd:
hdr = transaction_set.hdrFromFdno(f) hdr = transaction_set.hdrFromFdno(fd)
return hdr return hdr
def extract_rpm(name, path, target): def extract_rpm(name, path, target):
@ -116,8 +114,8 @@ def extract_rpm(name, path, target):
return rpm2cpio.wait() == 0 and cpio.wait() == 0 return rpm2cpio.wait() == 0 and cpio.wait() == 0
def parse_config(path): def parse_config(path):
with open(path, 'r') as f: with open(path, 'r') as fd:
return dict(line.rstrip('\n').split('=', 1) for line in f) return dict(line.rstrip('\n').split('=', 1) for line in fd)
def install(args, app): def install(args, app):
# TODO: Lock, mentioned in the note above # TODO: Lock, mentioned in the note above
@ -151,10 +149,10 @@ def install(args, app):
download(args, app, path_override=args.cachedir, download(args, app, path_override=args.cachedir,
dl_list=dl_list, suffix=UNVERIFIED_SUFFIX) dl_list=dl_list, suffix=UNVERIFIED_SUFFIX)
for idx, rpmfile in enumerate(rpm_list): for rpmfile in rpm_list:
path = rpmfile + UNVERIFIED_SUFFIX path = rpmfile + UNVERIFIED_SUFFIX
if not verify_rpm(path, args.nogpgcheck, transaction_set): if not verify_rpm(path, args.nogpgcheck, transaction_set):
parser.error('Package \'%s\' verification failed.' % template) parser.error('Package \'%s\' verification failed.' % rpmfile)
os.rename(path, rpmfile) os.rename(path, rpmfile)
for rpmfile in rpm_list: for rpmfile in rpm_list:
@ -208,44 +206,45 @@ def qrexec_popen(args, app, service, stdout=subprocess.PIPE, filter_esc=True):
service, service,
filter_esc=filter_esc, filter_esc=filter_esc,
stdout=stdout) stdout=stdout)
else: return subprocess.Popen([
return subprocess.Popen([ '/etc/qubes-rpc/%s' % service,
'/etc/qubes-rpc/%s' % service, ],
], stdin=subprocess.PIPE,
stdin=subprocess.PIPE, stdout=stdout,
stdout=stdout, stderr=subprocess.PIPE)
stderr=subprocess.PIPE)
def qrexec_payload(args, app, spec): def qrexec_payload(args, app, spec):
_ = app # unused
def check_newline(string, name): def check_newline(string, name):
if '\n' in string: if '\n' in string:
parser.error(f"Malformed {name}:" + parser.error(f"Malformed {name}:" +
" argument should not contain '\\n'.") " argument should not contain '\\n'.")
payload = '' payload = ''
for r in args.enablerepo if args.enablerepo else []: for repo in args.enablerepo if args.enablerepo else []:
check_newline(r, '--enablerepo') check_newline(repo, '--enablerepo')
payload += '--enablerepo=%s\n' % r payload += '--enablerepo=%s\n' % repo
for r in args.disablerepo if args.disablerepo else []: for repo in args.disablerepo if args.disablerepo else []:
check_newline(r, '--disablerepo') check_newline(repo, '--disablerepo')
payload += '--disablerepo=%s\n' % r payload += '--disablerepo=%s\n' % repo
for r in args.repoid if args.repoid else []: for repo in args.repoid if args.repoid else []:
check_newline(r, '--repoid') check_newline(repo, '--repoid')
payload += '--repoid=%s\n' % r payload += '--repoid=%s\n' % repo
check_newline(args.releasever, '--releasever') check_newline(args.releasever, '--releasever')
payload += '--releasever=%s\n' % args.releasever payload += '--releasever=%s\n' % args.releasever
check_newline(spec, 'template name') check_newline(spec, 'template name')
payload += spec + '\n' payload += spec + '\n'
payload += '---\n' payload += '---\n'
for fn in args.repo_files: for path in args.repo_files:
with open(fn, 'r') as f: with open(path, 'r') as fd:
payload += f.read() + '\n' payload += fd.read() + '\n'
return payload return payload
def qrexec_repoquery(args, app, spec='*'): def qrexec_repoquery(args, app, spec='*'):
proc = qrexec_popen(args, app, 'qubes.TemplateSearch') proc = qrexec_popen(args, app, 'qubes.TemplateSearch')
payload = qrexec_payload(args, app, spec) payload = qrexec_payload(args, app, spec)
stdout, stderr = proc.communicate(payload.encode('UTF-8')) stdout, _ = proc.communicate(payload.encode('UTF-8'))
stdout = stdout.decode('ASCII') stdout = stdout.decode('ASCII')
if proc.wait() != 0: if proc.wait() != 0:
raise ConnectionError("qrexec call 'qubes.TemplateSearch' failed.") raise ConnectionError("qrexec call 'qubes.TemplateSearch' failed.")
@ -259,18 +258,18 @@ def qrexec_repoquery(args, app, spec='*'):
return result return result
def qrexec_download(args, app, spec, path, dlsize=None): def qrexec_download(args, app, spec, path, dlsize=None):
with open(path, 'wb') as f: with open(path, 'wb') as fd:
# Don't filter ESCs for binary files # Don't filter ESCs for binary files
proc = qrexec_popen(args, app, 'qubes.TemplateDownload', proc = qrexec_popen(args, app, 'qubes.TemplateDownload',
stdout=f, filter_esc=False) stdout=fd, filter_esc=False)
payload = qrexec_payload(args, app, spec) payload = qrexec_payload(args, app, spec)
proc.stdin.write(payload.encode('UTF-8')) proc.stdin.write(payload.encode('UTF-8'))
proc.stdin.close() proc.stdin.close()
with tqdm.tqdm(desc=spec, total=dlsize, unit_scale=True, with tqdm.tqdm(desc=spec, total=dlsize, unit_scale=True,
unit_divisor=1000, unit='B') as pbar: unit_divisor=1000, unit='B') as pbar:
last = 0 last = 0
while proc.poll() == None: while proc.poll() is None:
cur = f.tell() cur = fd.tell()
pbar.update(cur - last) pbar.update(cur - last)
last = cur last = cur
time.sleep(0.1) time.sleep(0.1)
@ -304,6 +303,7 @@ def do_list(args, app):
(vm.name, version_str, TemplateState.INSTALLED.value)) (vm.name, version_str, TemplateState.INSTALLED.value))
if args.available or args.all: if args.available or args.all:
#pylint: disable=unused-variable
for name, epoch, version, release, reponame, dlsize, summary \ for name, epoch, version, release, reponame, dlsize, summary \
in query_res: in query_res:
version_str = build_version_str((epoch, version, release)) version_str = build_version_str((epoch, version, release))
@ -353,6 +353,7 @@ def get_dl_list(args, app):
parser.error('Package \'%s\' not found.' % template) parser.error('Package \'%s\' not found.' % template)
sys.exit(1) sys.exit(1)
# We only select one (latest) package for each distinct package name # We only select one (latest) package for each distinct package name
#pylint: disable=unused-variable
for name, epoch, version, release, reponame, dlsize, summary \ for name, epoch, version, release, reponame, dlsize, summary \
in query_res: in query_res:
ver = (epoch, version, release) ver = (epoch, version, release)
@ -364,7 +365,7 @@ def download(args, app, path_override=None, dl_list=None, suffix=''):
if dl_list is None: if dl_list is None:
dl_list = get_dl_list(args, app) dl_list = get_dl_list(args, app)
path = path_override if path_override != None else args.downloaddir path = path_override if path_override is not None else args.downloaddir
for name, (ver, dlsize) in dl_list.items(): for name, (ver, dlsize) in dl_list.items():
version_str = build_version_str(ver) version_str = build_version_str(ver)
spec = PACKAGE_NAME_PREFIX + name + '-' + version_str spec = PACKAGE_NAME_PREFIX + name + '-' + version_str
@ -380,27 +381,31 @@ def download(args, app, path_override=None, dl_list=None, suffix=''):
os.rename(target, target_suffix) os.rename(target, target_suffix)
else: else:
print('Downloading \'%s\'...' % spec, file=sys.stderr) print('Downloading \'%s\'...' % spec, file=sys.stderr)
ok = False done = False
for attempt in range(args.retries): for attempt in range(args.retries):
try: try:
qrexec_download(args, app, spec, target_suffix, dlsize) qrexec_download(args, app, spec, target_suffix, dlsize)
ok = True done = True
break break
except ConnectionError: except ConnectionError:
if attempt + 1 < args.retries: if attempt + 1 < args.retries:
print('\'%s\' download failed, retrying...' % spec, print('\'%s\' download failed, retrying...' % spec,
file=sys.stderr) file=sys.stderr)
if not ok: if not done:
print('\'%s\' download failed.' % spec, file=sys.stderr) print('\'%s\' download failed.' % spec, file=sys.stderr)
os.remove(target_suffix) os.remove(target_suffix)
sys.exit(1) sys.exit(1)
def remove(args, app): def remove(args, app):
_ = app # unused
# Use exec so stdio can be shared easily # Use exec so stdio can be shared easily
os.execvp('qvm-remove', ['qvm-remove'] + args.templates) os.execvp('qvm-remove', ['qvm-remove'] + args.templates)
def clean(args, app): def clean(args, app):
# TODO: More fine-grained options # TODO: More fine-grained options
_ = app # unused
shutil.rmtree(args.cachedir) shutil.rmtree(args.cachedir)
def main(args=None, app=None): def main(args=None, app=None):
@ -410,17 +415,17 @@ def main(args=None, app=None):
app = qubesadmin.Qubes() app = qubesadmin.Qubes()
if args.operation == 'install': if args.operation == 'install':
install(args, app) install(args, app)
elif args.operation == 'list': elif args.operation == 'list':
do_list(args, app) do_list(args, app)
elif args.operation == 'download': elif args.operation == 'download':
download(args, app) download(args, app)
elif args.operation == 'remove': elif args.operation == 'remove':
remove(args, app) remove(args, app)
elif args.operation == 'clean': elif args.operation == 'clean':
clean(args, app) clean(args, app)
else: else:
parser.error('Operation \'%s\' not supported.' % args.operation) parser.error('Operation \'%s\' not supported.' % args.operation)
return 0 return 0