Reject ISO downloads smaller than their declared volume extent
This commit is contained in:
Codex
+54
-13
@@ -47,25 +47,66 @@ class _SafeRedirect(urllib.request.HTTPRedirectHandler):
|
||||
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
||||
|
||||
|
||||
def _both_endian(field: bytes) -> int:
|
||||
half = len(field) // 2
|
||||
little = int.from_bytes(field[:half], 'little')
|
||||
if little != int.from_bytes(field[half:], 'big'):
|
||||
raise MediaError('Invalid ISO9660 descriptor: inconsistent byte-order fields')
|
||||
return little
|
||||
|
||||
|
||||
def _volume_extent(stream) -> None:
|
||||
"""Check ECMA-119 6.7.1 and PVD fields 8.4.8/8.4.12 against the file."""
|
||||
size = os.fstat(stream.fileno()).st_size
|
||||
extent = None
|
||||
udf = False
|
||||
terminated = False
|
||||
stream.seek(16 * 2048)
|
||||
for index in range(16, 257):
|
||||
sector = stream.read(2048)
|
||||
if len(sector) != 2048:
|
||||
break
|
||||
if sector[1:6] in (b'NSR02', b'NSR03') and sector[6] == 1:
|
||||
udf = True
|
||||
if sector[1:6] != b'CD001':
|
||||
continue
|
||||
if sector[0] == 1:
|
||||
if sector[6] != 1:
|
||||
raise MediaError('Unsupported ISO9660 primary volume descriptor version')
|
||||
blocks = _both_endian(sector[80:88])
|
||||
block_size = _both_endian(sector[128:132])
|
||||
# Raw ISO files use 2048-byte sectors. ECMA-119 6.2.2 permits
|
||||
# power-of-two logical blocks from 512 bytes up to sector size.
|
||||
if not blocks or block_size not in (512, 1024, 2048):
|
||||
raise MediaError('Invalid ISO9660 volume extent or logical block size')
|
||||
declared = blocks * block_size
|
||||
if declared > size:
|
||||
raise MediaError(f'ISO is truncated: declared volume extent is {declared} bytes, '
|
||||
f'but the image contains {size}; retry the direct ISO URL')
|
||||
if extent is not None and extent != declared:
|
||||
raise MediaError('Invalid ISO9660 descriptors: inconsistent volume extents')
|
||||
extent = declared
|
||||
elif sector[0] == 255 and sector[6] == 1:
|
||||
terminated = True
|
||||
if extent is not None and extent < (index + 1) * 2048:
|
||||
raise MediaError('Invalid ISO9660 volume extent: descriptor set lies outside the volume')
|
||||
break
|
||||
if extent is None:
|
||||
if udf:
|
||||
raise MediaError('Pure UDF image completeness is unsupported; use an ISO9660 or ISO9660/UDF hybrid ISO')
|
||||
raise MediaError('Not a supported ISO9660 image; use a direct ISO download URL rather than an HTML download page')
|
||||
if not terminated:
|
||||
raise MediaError('ISO9660 descriptor set is incomplete or unsupported; retry a complete ISO image')
|
||||
|
||||
|
||||
def validate_iso(path: Path, *, sha256: str | None = None) -> Path:
|
||||
"""Recognize ISO9660 or UDF volume descriptors and optional whole-file hash."""
|
||||
"""Validate ISO9660 extent (including hybrids) and optional whole-file hash."""
|
||||
wanted = _checksum(sha256)
|
||||
path = Path(path)
|
||||
if not path.is_file():
|
||||
raise MediaError(f'ISO does not exist or is not a file: {path}')
|
||||
found = False
|
||||
with path.open('rb') as stream:
|
||||
stream.seek(16 * 2048)
|
||||
for _ in range(241):
|
||||
sector = stream.read(2048)
|
||||
if len(sector) != 2048:
|
||||
break
|
||||
if ((sector[1:6] == b'CD001' and sector[0] in (0, 1, 2, 3, 255) and sector[6] == 1)
|
||||
or (sector[1:6] in (b'NSR02', b'NSR03') and sector[6] == 1)):
|
||||
found = True
|
||||
break
|
||||
if not found:
|
||||
raise MediaError('Not an ISO9660/UDF image; use a direct ISO download URL rather than an HTML download page')
|
||||
_volume_extent(stream)
|
||||
if wanted:
|
||||
stream.seek(0)
|
||||
digest = hashlib.file_digest(stream, 'sha256').hexdigest()
|
||||
|
||||
+82
-6
@@ -9,9 +9,18 @@ from unittest.mock import patch
|
||||
from hydra import media
|
||||
|
||||
|
||||
def iso_bytes(signature=b'CD001'):
|
||||
def iso_bytes(signature=b'CD001', *, declared_blocks=40, block_size=2048, hybrid=False):
|
||||
data = bytearray(40 * 2048)
|
||||
data[16 * 2048:16 * 2048 + 7] = b'\x01' + signature + b'\x01'
|
||||
start = 16 * 2048
|
||||
data[start:start + 7] = b'\x01' + signature + b'\x01'
|
||||
data[start + 80:start + 88] = declared_blocks.to_bytes(4, 'little') + declared_blocks.to_bytes(4, 'big')
|
||||
data[start + 120:start + 124] = b'\x01\x00\x00\x01'
|
||||
data[start + 124:start + 128] = b'\x01\x00\x00\x01'
|
||||
data[start + 128:start + 132] = block_size.to_bytes(2, 'little') + block_size.to_bytes(2, 'big')
|
||||
data[start + 881] = 1
|
||||
data[17 * 2048:17 * 2048 + 7] = b'\xffCD001\x01'
|
||||
if hybrid:
|
||||
data[20 * 2048:20 * 2048 + 7] = b'\x00NSR03\x01'
|
||||
return bytes(data)
|
||||
|
||||
|
||||
@@ -25,10 +34,12 @@ class MediaTests(unittest.TestCase):
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
owner.requests.append(self.path)
|
||||
body = b'<html>download here</html>' if self.path.startswith('/html') else iso_bytes()
|
||||
body = b'<html>download here</html>' if self.path.startswith('/html') else iso_bytes(
|
||||
declared_blocks=4000 if self.path.startswith('/extent') else 40)
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', 'text/html' if self.path.startswith('/html') else 'application/octet-stream')
|
||||
self.send_header('Content-Length', str(len(body) + (100 if self.path.startswith('/truncated') else 0)))
|
||||
if not self.path.startswith('/extent-no-length'):
|
||||
self.send_header('Content-Length', str(len(body) + (100 if self.path.startswith('/truncated') else 0)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
def log_message(self, *args):
|
||||
@@ -115,9 +126,9 @@ class MediaTests(unittest.TestCase):
|
||||
media.download_iso(self.url + '/os.iso', cache=self.cache)
|
||||
self.assertFalse(list(self.cache.glob('*.iso')))
|
||||
|
||||
def test_local_validation_accepts_udf_and_rejects_html_and_bad_checksum(self):
|
||||
def test_local_validation_accepts_iso_udf_hybrid_and_rejects_html_and_bad_checksum(self):
|
||||
path = Path(self.tmp.name) / 'os.iso'
|
||||
path.write_bytes(iso_bytes(b'NSR03'))
|
||||
path.write_bytes(iso_bytes(hybrid=True))
|
||||
media.validate_iso(path, sha256=hashlib.sha256(path.read_bytes()).hexdigest())
|
||||
with self.assertRaises(ValueError):
|
||||
media.validate_iso(path, sha256='0' * 64)
|
||||
@@ -132,3 +143,68 @@ class MediaTests(unittest.TestCase):
|
||||
self.assertEqual(media.download_iso(url, cache=self.cache, progress=lambda got, total: updates.append((got, total))), path)
|
||||
self.assertEqual(updates, [(81920, 81920)])
|
||||
self.assertEqual(self.requests, ['/cached.iso'])
|
||||
|
||||
def test_declared_extent_truncation_with_matching_content_length_never_publishes(self):
|
||||
with self.assertRaisesRegex(media.MediaError, 'truncated|extent'):
|
||||
media.download_iso(self.url + '/extent.iso', cache=self.cache)
|
||||
self.assertFalse(list(self.cache.glob('*.iso')))
|
||||
self.assertFalse(list(self.cache.glob('*.part')))
|
||||
|
||||
def test_declared_extent_truncation_without_content_length_never_publishes(self):
|
||||
with self.assertRaisesRegex(media.MediaError, 'truncated|extent'):
|
||||
media.download_iso(self.url + '/extent-no-length.iso', cache=self.cache)
|
||||
self.assertFalse(list(self.cache.glob('*.iso')))
|
||||
self.assertFalse(list(self.cache.glob('*.part')))
|
||||
|
||||
def test_zero_invalid_or_inconsistent_block_fields_are_rejected(self):
|
||||
path = Path(self.tmp.name) / 'invalid.iso'
|
||||
for body in (iso_bytes(declared_blocks=0), iso_bytes(block_size=0), iso_bytes(block_size=4096)):
|
||||
with self.subTest(body=body[16 * 2048 + 80:16 * 2048 + 132]):
|
||||
path.write_bytes(body)
|
||||
with self.assertRaises(media.MediaError):
|
||||
media.validate_iso(path)
|
||||
body = bytearray(iso_bytes())
|
||||
body[16 * 2048 + 84:16 * 2048 + 88] = (4000).to_bytes(4, 'big')
|
||||
path.write_bytes(body)
|
||||
with self.assertRaises(media.MediaError):
|
||||
media.validate_iso(path)
|
||||
body = bytearray(iso_bytes())
|
||||
body[16 * 2048 + 130:16 * 2048 + 132] = (1024).to_bytes(2, 'big')
|
||||
path.write_bytes(body)
|
||||
with self.assertRaises(media.MediaError):
|
||||
media.validate_iso(path)
|
||||
|
||||
def test_boot_signature_or_unterminated_primary_is_not_a_complete_image(self):
|
||||
path = Path(self.tmp.name) / 'descriptor.iso'
|
||||
boot_only = bytearray(iso_bytes())
|
||||
boot_only[16 * 2048] = 0
|
||||
unterminated = bytearray(iso_bytes())
|
||||
unterminated[17 * 2048:18 * 2048] = bytes(2048)
|
||||
for body in (boot_only, unterminated, iso_bytes()[:17 * 2048]):
|
||||
path.write_bytes(body)
|
||||
with self.assertRaises(media.MediaError):
|
||||
media.validate_iso(path)
|
||||
|
||||
def test_pure_udf_is_rejected_with_supported_image_guidance(self):
|
||||
path = Path(self.tmp.name) / 'udf.iso'
|
||||
for signature in (b'NSR02', b'NSR03'):
|
||||
path.write_bytes(iso_bytes(signature))
|
||||
with self.assertRaisesRegex(media.MediaError, 'UDF.*ISO9660|ISO9660.*UDF'):
|
||||
media.validate_iso(path)
|
||||
|
||||
def test_iso_udf_hybrid_with_trailing_data_and_smaller_blocks_remains_valid(self):
|
||||
path = Path(self.tmp.name) / 'hybrid.iso'
|
||||
for body in (iso_bytes(hybrid=True) + b'trailing hybrid partition data',
|
||||
iso_bytes(declared_blocks=80, block_size=1024),
|
||||
iso_bytes(declared_blocks=160, block_size=512)):
|
||||
path.write_bytes(body)
|
||||
self.assertEqual(media.validate_iso(path), path)
|
||||
|
||||
def test_cached_iso_truncated_with_descriptor_intact_is_not_reused(self):
|
||||
url = self.url + '/cached-truncated.iso'
|
||||
path = media.download_iso(url, cache=self.cache)
|
||||
path.write_bytes(path.read_bytes()[:20 * 2048])
|
||||
with self.assertRaisesRegex(media.MediaError, 'Cached ISO failed validation'):
|
||||
media.download_iso(url, cache=self.cache)
|
||||
self.assertEqual(path.stat().st_size, 20 * 2048)
|
||||
self.assertEqual(self.requests, ['/cached-truncated.iso'])
|
||||
|
||||
Reference in New Issue
Block a user