diff --git a/pycdlib/headervd.py b/pycdlib/headervd.py index e61606d4..4338976a 100644 --- a/pycdlib/headervd.py +++ b/pycdlib/headervd.py @@ -523,6 +523,12 @@ def add_rr_ce_entry(self, length): block = rockridge.RockRidgeContinuationBlock(0, self.log_block_size) self.rr_ce_blocks.append(block) offset = block.add_entry(length) + if offset is None: + # A brand new block had no room, so this entry is larger than a + # whole logical block. Callers split their entries into areas + # that each fit before getting here, so this means the caller + # got that wrong rather than that the ISO cannot be built. + raise pycdlibexception.PyCdlibInternalError('Rock Ridge Continuation Entry of length %d is too large to fit into a Continuation Block of size %d' % (length, self.log_block_size)) added_block = True return (added_block, block, offset) diff --git a/pycdlib/pycdlib.py b/pycdlib/pycdlib.py index d34b8c73..b591dc36 100644 --- a/pycdlib/pycdlib.py +++ b/pycdlib/pycdlib.py @@ -43,7 +43,7 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from typing import Any, BinaryIO, Callable, Deque, Dict, Generator, IO, List, Optional, Tuple, Union # noqa: F401 + from typing import Any, BinaryIO, Callable, Deque, Dict, Generator, IO, List, Optional, Set, Tuple, Union # noqa: F401 # There are a number of specific ways that numerical data is stored in the # ISO9660/Ecma-119 standard. In the text these are reference by the section @@ -363,11 +363,12 @@ def _reassign_vd_dirrecord_extents(vd, current_extent): file_list.append(dir_record.inode) if dir_record_rock_ridge is not None: - if dir_record_rock_ridge.dr_entries.ce_record is not None and dir_record_rock_ridge.ce_block is not None: - if dir_record_rock_ridge.ce_block.extent_location() < 0: - dir_record_rock_ridge.ce_block.set_extent_location(current_extent) - current_extent += 1 - dir_record_rock_ridge.dr_entries.ce_record.update_extent(dir_record_rock_ridge.ce_block.extent_location()) + if dir_record_rock_ridge.dr_entries.ce_record is not None and dir_record_rock_ridge.ce_areas: + for ce_area in dir_record_rock_ridge.ce_areas: + if ce_area.block.extent_location() < 0: + ce_area.block.set_extent_location(current_extent) + current_extent += 1 + dir_record_rock_ridge.dr_entries.ce_record.update_extent(dir_record_rock_ridge.ce_areas[0].extent_location()) if dir_record_rock_ridge.cl_to_moved_dr is not None: child_link_recs.append(dir_record) @@ -533,16 +534,15 @@ def _find_dr_record_by_name(vd, path, encoding): class PyCdlib: """The main class for manipulating ISOs.""" __slots__ = ('_initialized', '_cdfp', 'pvds', 'svds', 'vdsts', 'brs', 'pvd', - 'rock_ridge', '_always_consistent', '_has_udf', 'joliet_vd', - 'eltorito_boot_catalog', 'isohybrid_mbr', '_managing_fp', 'xa', - '_needs_reshuffle', '_rr_moved_record', '_rr_moved_name', - '_rr_moved_rr_name', 'enhanced_vd', 'version_vd', 'inodes', - 'interchange_level', '_write_check_list', '_track_writes', - 'udf_beas', 'udf_nsr', 'udf_teas', 'udf_anchors', - 'udf_main_descs', 'udf_reserve_descs', - 'udf_logical_volume_integrity', 'udf_boots', - 'udf_logical_volume_integrity_terminator', 'udf_root', - 'udf_file_set', 'udf_file_set_terminator', + 'rock_ridge', '_always_consistent', '_has_udf', 'joliet_vd', 'eltorito_boot_catalog', 'isohybrid_mbr', + '_managing_fp', 'xa', '_needs_reshuffle', '_rr_moved_record', + '_rr_moved_name', '_rr_moved_rr_name', 'enhanced_vd', + 'version_vd', 'inodes', 'interchange_level', + '_write_check_list', '_track_writes', 'udf_beas', 'udf_nsr', + 'udf_teas', 'udf_anchors', 'udf_main_descs', + 'udf_reserve_descs', 'udf_logical_volume_integrity', + 'udf_boots', 'udf_logical_volume_integrity_terminator', + 'udf_root', 'udf_file_set', 'udf_file_set_terminator', 'logical_block_size') def _initialize(self): @@ -1163,19 +1163,46 @@ def _walk_directories(self, vd, extent_to_ptr, extent_to_inode, rr_ce = '' if new_record.rock_ridge is not None and new_record.rock_ridge.dr_entries.ce_record is not None: - ce_record = new_record.rock_ridge.dr_entries.ce_record + ce_record = new_record.rock_ridge.dr_entries.ce_record # type: Optional[rockridge.RRCERecord] orig_pos = cdfp.tell() - self._seek_to_extent(ce_record.bl_cont_area) - cdfp.seek(ce_record.offset_cont_area, os.SEEK_CUR) - con_block = cdfp.read(ce_record.len_cont_area) - new_record.rock_ridge.parse(con_block, False, - new_record.rock_ridge.bytes_to_skip, - True, new_record.file_identifier()) + # A continuation area may itself end with a CE record + # pointing at a further area, chaining as many times as + # needed to hold the entries. Follow the whole chain, + # remembering where we have been so that an ISO whose CE + # records form a cycle cannot spin us forever. + seen_ce_areas = set() # type: Set[Tuple[int, int, int]] + num_ce_areas = 0 + while ce_record is not None: + area = (ce_record.bl_cont_area, + ce_record.offset_cont_area, + ce_record.len_cont_area) + if area in seen_ce_areas: + raise pycdlibexception.PyCdlibInvalidISO('Rock Ridge Continuation Entries form a loop') + seen_ce_areas.add(area) + num_ce_areas += 1 + + self._seek_to_extent(ce_record.bl_cont_area) + cdfp.seek(ce_record.offset_cont_area, os.SEEK_CUR) + con_block = cdfp.read(ce_record.len_cont_area) + new_record.rock_ridge.parse(con_block, False, + new_record.rock_ridge.bytes_to_skip, + True, new_record.file_identifier()) + block = self.pvd.track_rr_ce_entry(ce_record.bl_cont_area, + ce_record.offset_cont_area, + ce_record.len_cont_area) + new_record.rock_ridge.add_ce_area(block, + ce_record.offset_cont_area, + ce_record.len_cont_area) + + # Parsing an area stores any CE it contained in + # ce_entries; take it as the next link and clear it so + # the following area can carry one of its own, and so + # that no stale link is left behind at the end. + ce_entries = new_record.rock_ridge.ce_entries + ce_record = ce_entries.ce_record if ce_entries is not None else None + if ce_entries is not None: + ce_entries.ce_record = None cdfp.seek(orig_pos) - block = self.pvd.track_rr_ce_entry(ce_record.bl_cont_area, - ce_record.offset_cont_area, - ce_record.len_cont_area) - new_record.rock_ridge.update_ce_block(block) rr_ce = new_record.rock_ridge.rr_version if new_record.rock_ridge else '' # The PX record could be in the continuation blob, so @@ -2787,12 +2814,35 @@ def _write_directory_records(self, vd, outfp, progress): if child.rock_ridge is not None: if child.rock_ridge.dr_entries.ce_record is not None: - # The child has a continue block, so write it out here. - ce_rec = child.rock_ridge.dr_entries.ce_record - outfp.seek(ce_rec.bl_cont_area * self.logical_block_size + ce_rec.offset_cont_area) - rec = child.rock_ridge.record_ce_entries() - self._outfp_write_with_check(outfp, rec) - progress.call(len(rec)) + # The child has continuation areas, so write them out + # here. There is usually just the one, but where the + # entries do not fit they are chained across several. + ce_areas = child.rock_ridge.ce_areas + if ce_areas: + ce_rec = child.rock_ridge.dr_entries.ce_record + for ce_area, rec in zip(ce_areas, child.rock_ridge.record_ce_areas()): + extent = ce_area.extent_location() + if extent < 0: + # Reshuffling deliberately skips the dot and + # dotdot records, so their Continuation + # Blocks never get an extent assigned and + # they keep the one they were parsed with. + # Their entries are small and always fit in + # a single area, so there is no chain here. + extent = ce_rec.bl_cont_area + outfp.seek(extent * self.logical_block_size + ce_area.offset) + self._outfp_write_with_check(outfp, rec) + progress.call(len(rec)) + else: + # The Rock Ridge 'ER' area of the root gets an + # extent of its own rather than a slot in a + # Continuation Block, so it has no area tracked + # against it; it is always small enough for one. + ce_rec = child.rock_ridge.dr_entries.ce_record + outfp.seek(ce_rec.bl_cont_area * self.logical_block_size + ce_rec.offset_cont_area) + rec = child.rock_ridge.record_ce_entries() + self._outfp_write_with_check(outfp, rec) + progress.call(len(rec)) if child.rock_ridge.child_link_record_exists(): continue @@ -3076,12 +3126,25 @@ def _update_rr_ce_entry(self, rec): The number of additional bytes needed for this Rock Ridge CE entry. """ if rec.rock_ridge is not None and rec.rock_ridge.dr_entries.ce_record is not None: - celen = rec.rock_ridge.dr_entries.ce_record.len_cont_area - added_block, block, offset = self.pvd.add_rr_ce_entry(celen) - rec.rock_ridge.update_ce_block(block) - rec.rock_ridge.dr_entries.ce_record.update_offset(offset) - if added_block: - return self.logical_block_size + # The entries may need more than one area to hold them, in which + # case each area but the last ends with a CE record linking to the + # next. Allocate them all; they need not be adjacent, or even in + # the same Continuation Block. + rec.rock_ridge.clear_ce_areas() + num_bytes_to_add = 0 + for celen in rec.rock_ridge.ce_area_lengths(self.logical_block_size): + added_block, block, offset = self.pvd.add_rr_ce_entry(celen) + rec.rock_ridge.add_ce_area(block, offset, celen) + if added_block: + num_bytes_to_add += self.logical_block_size + + # The CE record in the directory record describes the first area + # only; the rest are reached by following the chain. + first = rec.rock_ridge.ce_areas[0] + rec.rock_ridge.dr_entries.ce_record.update_offset(first.offset) + rec.rock_ridge.dr_entries.ce_record.update_len(first.length) + + return num_bytes_to_add return 0 @@ -5446,9 +5509,9 @@ def rm_directory(self, iso_path=None, rr_name=None, joliet_path=None, # pylint: # child_link record because it is a 'fake' record that has no # size. - if child.rock_ridge is not None and child.rock_ridge.dr_entries.ce_record is not None and child.rock_ridge.ce_block is not None: - child.rock_ridge.ce_block.remove_entry(child.rock_ridge.dr_entries.ce_record.offset_cont_area, - child.rock_ridge.dr_entries.ce_record.len_cont_area) + if child.rock_ridge is not None and child.rock_ridge.dr_entries.ce_record is not None: + for ce_area in child.rock_ridge.ce_areas: + ce_area.block.remove_entry(ce_area.offset, ce_area.length) if joliet_path is not None: num_bytes_to_remove += self._rm_joliet_dir(self._normalize_joliet_path(joliet_path)) diff --git a/pycdlib/rockridge.py b/pycdlib/rockridge.py index bf0ab07b..6c3e16dd 100644 --- a/pycdlib/rockridge.py +++ b/pycdlib/rockridge.py @@ -65,7 +65,6 @@ _RR_SINGLE_INSTANCE_FIELDS = { b'SP': operator.attrgetter('sp_record'), b'RR': operator.attrgetter('rr_record'), - b'CE': operator.attrgetter('ce_record'), b'PX': operator.attrgetter('px_record'), b'ST': operator.attrgetter('st_record'), b'ER': operator.attrgetter('er_record'), @@ -383,6 +382,21 @@ def update_offset(self, offset): self.offset_cont_area = offset + def update_len(self, length): + # type: (int) -> None + """ + Set the length of the continuation area this CE record points at. + + Parameters: + length - The new length of the continuation area. + Returns: + Nothing. + """ + if not self._initialized: + raise pycdlibexception.PyCdlibInternalError('CE record not initialized') + + self.len_cont_area = length + def add_record(self, length): # type: (int) -> None """ @@ -2584,10 +2598,37 @@ def __init__(self): # than 8.3. Rock Ridge depends on the System Use and Sharing Protocol (SUSP), # which defines some standards on how to use the System Area. +class RockRidgeContinuationArea: + """ + A class representing one area holding Rock Ridge Continuation entries. + Where the entries do not all fit into a single area, each area but the last + ends with a CE record linking to the next one. + """ + __slots__ = ('block', 'offset', 'length') + + def __init__(self, block, offset, length): + # type: (RockRidgeContinuationBlock, int, int) -> None + self.block = block + self.offset = offset + self.length = length + + def extent_location(self): + # type: () -> int + """ + Get the extent location of the block this area lives in. + + Parameters: + None. + Returns: + The extent location of the block this area lives in. + """ + return self.block.extent_location() + + class RockRidge: """A class representing Rock Ridge entries.""" __slots__ = ('_initialized', 'dr_entries', 'ce_entries', 'cl_to_moved_dr', - 'moved_to_cl_dr', 'parent_link', 'rr_version', 'ce_block', + 'moved_to_cl_dr', 'parent_link', 'rr_version', 'ce_areas', 'bytes_to_skip', '_full_name') def __init__(self): @@ -2604,7 +2645,9 @@ def __init__(self): self.moved_to_cl_dr = None # type: Optional[dr.DirectoryRecord] self.parent_link = None # type: Optional[dr.DirectoryRecord] self.rr_version = '' - self.ce_block = None # type: Optional[RockRidgeContinuationBlock] + # The continuation entries may be split across several areas, each + # linked to the next by a CE record; see add_ce_area(). + self.ce_areas = [] # type: List[RockRidgeContinuationArea] self._initialized = False def _ensure_ce_entries(self): @@ -2673,6 +2716,11 @@ def parse(self, record, is_first_dir_record_of_root, bytes_to_skip, has_es_record = False sf_record_length = None er_id = None + # A CE record is single-instance per System Use area rather than per + # RockRidge object: an area may chain to a further area, and each link + # in that chain gets its own CE. Since parse() is called once per + # area, a local flag enforces exactly the right invariant. + seen_ce = False while True: if left == 0: break @@ -2711,6 +2759,9 @@ def parse(self, record, is_first_dir_record_of_root, bytes_to_skip, entry_list.rr_record = RRRRRecord() entry_list.rr_record.parse(recslice) elif rtype == b'CE': + if seen_ce: + raise pycdlibexception.PyCdlibInvalidISO('Only single CE record supported') + seen_ce = True entry_list.ce_record = RRCERecord() entry_list.ce_record.parse(recslice) elif rtype == b'PX': @@ -2834,6 +2885,20 @@ def _record(self, entries): Returns: A string representing the Rock Ridge entry. """ + return b''.join(self._record_list(entries)) + + def _record_list(self, entries): + # type: (RockRidgeEntries) -> List[bytes] + """ + Return the individual SUSP records making up a Rock Ridge entry. The + continuation area splitter needs the records one at a time, since an + area may only be cut on a record boundary. + + Parameters: + entries - The dr_entries or ce_entries to generate records for. + Returns: + A list of strings, one per SUSP record. + """ outlist = [] if entries.sp_record is not None: @@ -2884,7 +2949,7 @@ def _record(self, entries): if entries.sf_record is not None: outlist.append(entries.sf_record.record()) - return b''.join(outlist) + return outlist def record_dr_entries(self): # type: () -> bytes @@ -3776,20 +3841,134 @@ def relocated_record(self): return True return self.ce_entries is not None and self.ce_entries.re_record is not None - def update_ce_block(self, block): - # type: (RockRidgeContinuationBlock) -> None + def add_ce_area(self, block, offset, length): + # type: (RockRidgeContinuationBlock, int, int) -> None """ - Update the Continuation Entry block object used by this Rock Ridge Record. + Add an area holding some of the Continuation entries for this record. + Where the entries need more than one area, they are added in the order + they are chained together on the ISO. Parameters: - block - The new block object. + block - The block object the area lives in. + offset - The offset within the block that the area starts at. + length - The length of the area, including any linking CE record. Returns: Nothing. """ if not self._initialized: raise pycdlibexception.PyCdlibInternalError('Rock Ridge extension not initialized') - self.ce_block = block + self.ce_areas.append(RockRidgeContinuationArea(block, offset, length)) + + def clear_ce_areas(self): + # type: () -> None + """ + Forget the areas holding the Continuation entries for this record, so + that they can be allocated afresh. + + Parameters: + None. + Returns: + Nothing. + """ + if not self._initialized: + raise pycdlibexception.PyCdlibInternalError('Rock Ridge extension not initialized') + + self.ce_areas = [] + + def ce_area_lengths(self, max_area_size): + # type: (int) -> List[int] + """ + Work out how to divide the Continuation entries into areas of no more + than max_area_size bytes. An area may only be cut on a SUSP record + boundary, and every area but the last has to leave room for the CE + record linking it to the next one. No SUSP record can be longer than + 255 bytes, so a cut point always exists. + + Parameters: + max_area_size - The largest an individual area may be. + Returns: + A list of area lengths, each including its linking CE record. + """ + if not self._initialized: + raise pycdlibexception.PyCdlibInternalError('Rock Ridge extension not initialized') + + if self.ce_entries is None: + return [] + + lengths = [len(rec) for rec in self._record_list(self.ce_entries)] + if not lengths: + return [] + + total = sum(lengths) + if total <= max_area_size: + # Everything fits in a single area, so no linking is needed. + return [total] + + budget = max_area_size - RRCERecord.length() + areas = [] + used = 0 + for length in lengths: + if length > budget: + raise pycdlibexception.PyCdlibInternalError('Rock Ridge entry is too large to fit into a Continuation Area') + if used + length > budget: + areas.append(used + RRCERecord.length()) + used = 0 + used += length + areas.append(used) + + return areas + + def record_ce_areas(self): + # type: () -> List[bytes] + """ + Return the contents of each area holding this record's Continuation + entries, in the order the areas are chained together. Each area but + the last ends with a CE record pointing at the one after it, so the + areas must already have been assigned locations. + + Parameters: + None. + Returns: + A list of strings, one per continuation area. + """ + if not self._initialized: + raise pycdlibexception.PyCdlibInternalError('Rock Ridge extension not initialized') + + if self.ce_entries is None or not self.ce_areas: + return [] + + recs = self._record_list(self.ce_entries) + index = 0 + outlist = [] + for areanum, area in enumerate(self.ce_areas): + last = areanum == (len(self.ce_areas) - 1) + budget = area.length + if not last: + budget -= RRCERecord.length() + + used = 0 + chunk = [] + while index < len(recs) and (used + len(recs[index])) <= budget: + used += len(recs[index]) + chunk.append(recs[index]) + index += 1 + + if not last: + nxt = self.ce_areas[areanum + 1] + ce_record = RRCERecord() + ce_record.new() + ce_record.update_extent(nxt.extent_location()) + ce_record.update_offset(nxt.offset) + ce_record.add_record(nxt.length) + chunk.append(ce_record.record()) + + outlist.append(b''.join(chunk)) + + if index != len(recs): + raise pycdlibexception.PyCdlibInternalError('Rock Ridge Continuation entries do not fit into the areas allocated for them') + + return outlist class RockRidgeContinuationEntry: @@ -3908,7 +4087,7 @@ def track_entry(self, offset, length): bisect.insort_left(self._entries, RockRidgeContinuationEntry(offset, length)) def add_entry(self, length): - # type: (int) -> int + # type: (int) -> Optional[int] """ Add a new entry to this Rock Ridge Continuation Block. This method attempts to find a gap that fits the new length anywhere within this @@ -3949,9 +4128,11 @@ def add_entry(self, length): if self._max_block_size >= length: offset = 0 - if offset >= 0: - bisect.insort_left(self._entries, - RockRidgeContinuationEntry(offset, length)) + if offset < 0: + return None + + bisect.insort_left(self._entries, + RockRidgeContinuationEntry(offset, length)) return offset diff --git a/tests/integration/test_new.py b/tests/integration/test_new.py index b8522a2c..362976e1 100644 --- a/tests/integration/test_new.py +++ b/tests/integration/test_new.py @@ -8403,6 +8403,64 @@ def test_new_rr_empty_dir_get_record(): iso.close() +def test_new_rr_long_names_overflow_ce_block(): + # Regression test for issue #177: enough files with long Rock Ridge names + # to overflow a single Rock Ridge Continuation Block. Once the first + # block filled up, add_entry() returned -1 instead of None, which + # add_rr_ce_entry() accepted as a valid offset instead of allocating a + # second block. The bogus -1 offset then failed on write. + iso = pycdlib.PyCdlib() + iso.new(rock_ridge='1.09') + + for i in range(40): + rr_name = ('f%03d' % i) + 'x' * 200 + iso.add_fp(io.BytesIO(b'hello\n'), 6, '/FILE%03d.;1' % i, rr_name=rr_name) + + out = io.BytesIO() + iso.write_fp(out) + + iso.close() + + # Make sure the result round-trips and the long names survived. + iso2 = pycdlib.PyCdlib() + iso2.open_fp(out) + for i in range(40): + rr_name = ('f%03d' % i) + 'x' * 200 + rec = iso2.get_record(rr_path='/' + rr_name) + assert(rec.get_data_length() == 6) + iso2.close() + +def test_new_rr_symlink_chained_ce(): + # A symlink whose target needs more continuation area than fits in a single + # logical block gets split across several areas, each linking to the next + # with a CE record. + iso = pycdlib.PyCdlib() + iso.new(rock_ridge='1.09') + + # 4019 bytes, with every component well under NAME_MAX (255) and the whole + # target under PATH_MAX (4096), so this is a symlink that can really exist + # on a Unix filesystem. Anything past roughly a 2040-byte target needs + # more than one 2048-byte block for its continuation area. + target = '/'.join(['c' * 200] * 20) + iso.add_symlink('/SYM.;1', 'sym', target) + + rec = iso.get_record(rr_path='/sym') + assert(len(rec.rock_ridge.ce_areas) > 1) + for ce_area in rec.rock_ridge.ce_areas: + assert(ce_area.length <= 2048) + + out = io.BytesIO() + iso.write_fp(out) + iso.close() + + # Now make sure it reads back as the same symlink. + iso2 = pycdlib.PyCdlib() + iso2.open_fp(out) + rec2 = iso2.get_record(rr_path='/sym') + assert(len(rec2.rock_ridge.ce_areas) > 1) + assert(rec2.rock_ridge.symlink_path() == target.encode('utf-8')) + iso2.close() + def test_new_isolevel4_deep_directory(): iso = pycdlib.PyCdlib() iso.new(interchange_level=4) diff --git a/tests/integration/test_parse.py b/tests/integration/test_parse.py index a098890a..b320be9b 100644 --- a/tests/integration/test_parse.py +++ b/tests/integration/test_parse.py @@ -10,6 +10,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) import pycdlib +import pycdlib.rockridge from test_common import * @@ -3290,3 +3291,91 @@ def test_parse_one_extent_path_tables(tmpdir): outfp.write(shrunk) do_a_test(tmpdir, outfile, check_onefile_one_extent_path_tables) + +def _write_chained_ce_iso(path): + """ + Write an ISO holding a symlink whose target needs more than one Rock Ridge + continuation area, so that the areas have to be chained together. Returns + the symlink target. + """ + target = '/'.join(['c' * 200] * 20) + + iso = pycdlib.PyCdlib() + iso.new(rock_ridge='1.09') + iso.add_symlink('/LINK.;1', 'link', target) + iso.write(path) + iso.close() + + return target + +def test_parse_rr_chained_ce(tmpdir): + # A symlink target too long to fit in a single continuation area is spread + # across several, each linking to the next with a CE record. + outfile = str(tmpdir.join('chainedce.iso')) + target = _write_chained_ce_iso(outfile) + + iso = pycdlib.PyCdlib() + iso.open(outfile) + rec = iso.get_record(rr_path='/link') + assert(len(rec.rock_ridge.ce_areas) > 1) + for ce_area in rec.rock_ridge.ce_areas: + assert(ce_area.length <= 2048) + assert(rec.rock_ridge.symlink_path() == target.encode('utf-8')) + iso.close() + +def test_parse_rr_chained_ce_round_trip(tmpdir): + # Reading an ISO with chained continuation areas and writing it back out + # has to reproduce it exactly. + first = str(tmpdir.join('chainedce.iso')) + second = str(tmpdir.join('chainedce2.iso')) + target = _write_chained_ce_iso(first) + + iso = pycdlib.PyCdlib() + iso.open(first) + iso.write(second) + iso.close() + + iso = pycdlib.PyCdlib() + iso.open(second) + assert(iso.get_record(rr_path='/link').rock_ridge.symlink_path() == target.encode('utf-8')) + iso.close() + + with open(first, 'rb') as infp: + firstdata = infp.read() + with open(second, 'rb') as infp: + seconddata = infp.read() + assert(firstdata == seconddata) + +def test_parse_rr_ce_loop(tmpdir): + # A CE record pointing back at the area holding it would spin the parser + # forever without a loop guard. pycdlib will not write one of these, so + # take a good ISO and corrupt the link. + outfile = str(tmpdir.join('celoop.iso')) + _write_chained_ce_iso(outfile) + + iso = pycdlib.PyCdlib() + iso.open(outfile) + first_area = iso.get_record(rr_path='/link').rock_ridge.ce_areas[0] + extent = first_area.extent_location() + offset = first_area.offset + length = first_area.length + iso.close() + + ce_record = pycdlib.rockridge.RRCERecord() + ce_record.new() + ce_record.update_extent(extent) + ce_record.update_offset(offset) + ce_record.update_len(length) + + # The CE linking the first area to the second sits at the end of the first. + with open(outfile, 'rb') as infp: + data = bytearray(infp.read()) + link = extent * 2048 + offset + length - 28 + data[link:link + 28] = ce_record.record() + with open(outfile, 'wb') as outfp: + outfp.write(bytes(data)) + + iso = pycdlib.PyCdlib() + with pytest.raises(pycdlib.pycdlibexception.PyCdlibInvalidISO) as excinfo: + iso.open(outfile) + assert(str(excinfo.value) == 'Rock Ridge Continuation Entries form a loop') diff --git a/tests/unit/test_rockridge.py b/tests/unit/test_rockridge.py index 2b52a738..8c4a0761 100644 --- a/tests/unit/test_rockridge.py +++ b/tests/unit/test_rockridge.py @@ -205,6 +205,12 @@ def test_rrcerecord_update_offset_not_initialized(): ce.update_offset(0) assert(str(excinfo.value) == 'CE record not initialized') +def test_rrcerecord_update_len_not_initialized(): + ce = pycdlib.rockridge.RRCERecord() + with pytest.raises(pycdlib.pycdlibexception.PyCdlibInternalError) as excinfo: + ce.update_len(0) + assert(str(excinfo.value) == 'CE record not initialized') + def test_rrcerecord_update_add_record_not_initialized(): ce = pycdlib.rockridge.RRCERecord() with pytest.raises(pycdlib.pycdlibexception.PyCdlibInternalError) as excinfo: @@ -1528,10 +1534,28 @@ def test_rr_relocated_record_not_initialized(): rr.relocated_record() assert(str(excinfo.value) == 'Rock Ridge extension not initialized') -def test_rr_update_ce_block_not_initialized(): +def test_rr_add_ce_area_not_initialized(): rr = pycdlib.rockridge.RockRidge() with pytest.raises(pycdlib.pycdlibexception.PyCdlibInternalError) as excinfo: - rr.update_ce_block(None) + rr.add_ce_area(None, 0, 0) + assert(str(excinfo.value) == 'Rock Ridge extension not initialized') + +def test_rr_clear_ce_areas_not_initialized(): + rr = pycdlib.rockridge.RockRidge() + with pytest.raises(pycdlib.pycdlibexception.PyCdlibInternalError) as excinfo: + rr.clear_ce_areas() + assert(str(excinfo.value) == 'Rock Ridge extension not initialized') + +def test_rr_ce_area_lengths_not_initialized(): + rr = pycdlib.rockridge.RockRidge() + with pytest.raises(pycdlib.pycdlibexception.PyCdlibInternalError) as excinfo: + rr.ce_area_lengths(2048) + assert(str(excinfo.value) == 'Rock Ridge extension not initialized') + +def test_rr_record_ce_areas_not_initialized(): + rr = pycdlib.rockridge.RockRidge() + with pytest.raises(pycdlib.pycdlibexception.PyCdlibInternalError) as excinfo: + rr.record_ce_areas() assert(str(excinfo.value) == 'Rock Ridge extension not initialized') def test_rr_parse_continuation_does_not_downgrade_version(): @@ -1706,6 +1730,63 @@ def test_rrcontentry_add_multiple(): assert(rr._entries[2].offset == 40) assert(rr._entries[2].length == 12) +def _ce_record(block, offset, length): + def swab(x): + return struct.unpack('>I', struct.pack('