Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions src/bmaptool/BmapCopy.py
Original file line number Diff line number Diff line change
Expand Up @@ -403,8 +403,14 @@ def _parse_bmap(self):
self._f_bmap.seek(0)
xml_extract = ""
for num, line in enumerate(self._f_bmap):
if num >= err.position[0] - 4 and num <= err.position[0] + 4:
xml_extract += "Line %d: %s" % (num, line)
if num < err.position[0] - 4:
continue
if num > err.position[0] + 4:
break
xml_extract += "Line %d: %s" % (
num,
line.decode(errors="backslashreplace"),
)

raise Error(
"cannot parse the bmap file '%s' which should be a "
Expand Down Expand Up @@ -715,7 +721,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1]

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
4 changes: 2 additions & 2 deletions src/bmaptool/BmapHelpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,8 @@ def human_size(size):

def human_time(seconds):
"""Transform time in seconds to the HH:MM:SS format."""
(minutes, seconds) = divmod(seconds, 60)
(hours, minutes) = divmod(minutes, 60)
minutes, seconds = divmod(seconds, 60)
hours, minutes = divmod(minutes, 60)

result = ""
if hours:
Expand Down
36 changes: 34 additions & 2 deletions src/bmaptool/CLI.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,9 @@ def __init__(self, file_obj, name):
self._file_obj = file_obj
self.name = name

def __iter__(self):
return iter(self._file_obj)

def __getattr__(self, name):
return getattr(self._file_obj, name)

Expand Down Expand Up @@ -215,7 +218,7 @@ def verify_bmap_signature_gpgbin(bmap_obj, detached_sig, gpgargv, keyring):
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
(output, error) = sp.communicate()
output, error = sp.communicate()
if sp.returncode > 0:
if error.find(b"[GNUPG:] NO_PUBKEY "):
error_out("No matching key found")
Expand Down Expand Up @@ -362,6 +365,35 @@ def _add_ext(p, ext):
have_method.add("gpgv")

if not have_method:
# if no method is available: that is okay i --no-sig-verify was passed
if args.no_sig_verify:
# if signature was detached, there is no cleartext to return
if detached_sig:
return None
# Since no gpg is available and no signature verification is
# required, strip off the signature from the plaintext manually.
# Since mmap is used later, we cannot use io.BytesIO() but need a
# real file.
try:
tmp_obj = tempfile.TemporaryFile("w+b")
except IOError as err:
error_out("cannot create a temporary file for bmap:\n%s", err)

header_done = False
for line in bmap_obj.readlines():
if not header_done:
# header is done after the first empty line
if line == b"\n":
header_done = True
continue
if line == b"-----BEGIN PGP SIGNATURE-----\n":
break
if line.endswith(b"-----BEGIN PGP SIGNATURE-----\n"):
line.removeprefix(b"- ")
tmp_obj.write(line)

tmp_obj.seek(0)
return tmp_obj
error_out("Cannot verify GPG signature without GPG")

for method in ["gpgme", "gpgv", "gpg"]:
Expand Down Expand Up @@ -510,7 +542,7 @@ def open_files(args):

# Open the bmap file. Try to discover the bmap file automatically if it
# was not specified.
(bmap_obj, bmap_path) = find_and_open_bmap(args, image_obj.is_url)
bmap_obj, bmap_path = find_and_open_bmap(args, image_obj.is_url)

if bmap_path == args.image:
# Most probably the user specified the bmap file instead of the image
Expand Down
8 changes: 6 additions & 2 deletions src/bmaptool/TransRead.py
Original file line number Diff line number Diff line change
Expand Up @@ -222,9 +222,13 @@ def __del__(self):

if getattr(self, "_child_processes"):
for child in self._child_processes:
if child.poll() is None:
returncode = child.poll()
if returncode is None:
# process is still running
child.kill()
child.wait()
returncode = child.wait()
if returncode != 0:
raise Error(f"non-zero exit of {child.args}: {returncode}")
self._child_processes = []

def _read_thread(self, f_from, f_to):
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy1_0.py
Original file line number Diff line number Diff line change
Expand Up @@ -501,7 +501,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_0.py
Original file line number Diff line number Diff line change
Expand Up @@ -450,7 +450,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_1.py
Original file line number Diff line number Diff line change
Expand Up @@ -449,7 +449,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_2.py
Original file line number Diff line number Diff line change
Expand Up @@ -450,7 +450,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_3.py
Original file line number Diff line number Diff line change
Expand Up @@ -486,7 +486,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_4.py
Original file line number Diff line number Diff line change
Expand Up @@ -486,7 +486,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_5.py
Original file line number Diff line number Diff line change
Expand Up @@ -526,7 +526,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy2_6.py
Original file line number Diff line number Diff line change
Expand Up @@ -526,7 +526,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
2 changes: 1 addition & 1 deletion tests/oldcodebase/BmapCopy3_0.py
Original file line number Diff line number Diff line change
Expand Up @@ -575,7 +575,7 @@ def copy(self, sync=True, verify=True):
exc_info = batch[1]
raise exc_info[1].with_traceback(exc_info[2])

(start, end, buf) = batch[1:4]
start, end, buf = batch[1:4]

assert len(buf) <= (end - start + 1) * self.block_size
assert len(buf) > (end - start) * self.block_size
Expand Down
1 change: 0 additions & 1 deletion tests/test_bmap_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
from backports.tempfile import TemporaryDirectory
from bmaptool import BmapHelpers


# This is a work-around for Centos 6
try:
import unittest2 as unittest # pylint: disable=F0401
Expand Down
Loading