Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import binascii
2import io
3from pathlib import Path
4from typing import BinaryIO
6from unblob.file_utils import (
7 Endian,
8 FileSystem,
9 InvalidInputFormat,
10 StructParser,
11 iterate_file,
12)
13from unblob.models import (
14 Extractor,
15 ExtractResult,
16 File,
17 HandlerDoc,
18 HandlerType,
19 HexString,
20 Reference,
21 StructHandler,
22 ValidChunk,
23)
24from unblob.report import ExtractionProblem
26C_DEFINITION = r"""
27 typedef struct lvm_label_header {
28 char signature[8]; // "LABELONE"
29 uint64 sector; // sector number of this label header, usually 1
30 uint32 crc; // CRC of fields below to end of sector
31 uint32 header_size; // size of this header; pv_header follows immediately
32 char type[8]; // "LVM2 001"
33 } lvm_label_header_t;
35 typedef struct lvm_pv_header {
36 char uuid[32]; // PV UUID, ASCII
37 uint64 device_size; // PV size in bytes
38 } lvm_pv_header_t;
40 typedef struct lvm_data_area_descriptor {
41 uint64 area_offset; // relative to start of the PV
42 uint64 area_size; // 0 = unbounded
43 } lvm_data_area_descriptor_t;
45 typedef struct lvm_raw_location_descriptor {
46 uint64 data_offset; // relative to start of metadata area
47 uint64 data_size;
48 uint32 crc;
49 uint32 flags; // 0x1 = ignored
50 } lvm_raw_location_descriptor_t;
52 typedef struct lvm_metadata_area_header {
53 uint32 crc;
54 char magic[16]; // "\x20LVM2\x20x[5A%r0N*>" signature
55 uint32 version;
56 uint64 offset; // metadata area offset from PV start
57 uint64 size; // metadata area size
58 lvm_raw_location_descriptor_t locns[4];
59 char padding[376];
60 } lvm_metadata_area_header_t;
62 """
64SECTOR_SIZE = 512 # LVM2 format constant
65LABEL_SCAN_SECTORS = 4
66LABEL_HEADER_SIZE = 32
67PV_HEADER_SIZE = 40
68DESCRIPTOR_SIZE = 16
69MDA_MAGIC = b" LVM2 x[5A%r0N*>"
70MDA_VERSION = 1
71INITIAL_CRC = 0xF597A6CF
72RAW_LOCATION_IGNORED = 0x1
75def _lvm_crc32(data: bytes | bytearray) -> int:
76 """Calculate LVM's non-inverted, seeded CRC-32."""
77 return (binascii.crc32(data, INITIAL_CRC ^ 0xFFFFFFFF) ^ 0xFFFFFFFF) & 0xFFFFFFFF
80def parse_lvm_metadata(text: str) -> dict:
81 """Parse LVM2 text metadata into nested dicts.
83 Grammar (per libvslvm §5.5):
84 section { ... } section opens a named scope
85 key = value int, "string", or [list]
86 # ... comment to end of line
87 Lists may span multiple lines until ']'.
88 """
89 root: dict = {}
90 stack: list[dict] = [root]
91 lines = _clean(text)
93 for line in lines:
94 if line == "}":
95 if len(stack) == 1:
96 raise InvalidInputFormat("Unexpected closing brace in LVM metadata.")
97 stack.pop()
98 elif line.endswith("{"):
99 section: dict = {}
100 stack[-1][line[:-1].strip()] = section
101 stack.append(section)
102 elif "=" in line:
103 key, _, value = (s.strip() for s in line.partition("="))
104 if value.startswith("[") and "]" not in value:
105 value = _consume_list(value, lines)
106 stack[-1][key] = _parse_value(value)
108 if len(stack) != 1:
109 raise InvalidInputFormat("Unclosed section in LVM metadata.")
110 return root
113def _clean(text: str):
114 """Yield non-empty, comment-stripped lines."""
115 for raw in text.splitlines():
116 line = raw.split("#", 1)[0].strip()
117 if line:
118 yield line
121def _consume_list(first: str, lines) -> str:
122 """Join continuation lines until ']' is seen, return the full list value as one string."""
123 parts = [first]
124 for line in lines:
125 parts.append(line)
126 if "]" in line:
127 return " ".join(parts)
128 raise InvalidInputFormat("Unclosed list in LVM metadata.")
131def _parse_value(text: str) -> int | str | list:
132 if text.startswith('"') and text.endswith('"'):
133 return text[1:-1]
134 if text.startswith("[") and text.endswith("]"):
135 items = [chunk.strip() for chunk in text[1:-1].split(",")]
136 return [_parse_value(item) for item in items if item]
137 if text.lstrip("-").isdigit():
138 return int(text)
139 return text
142class LVM2Extractor(Extractor):
143 def __init__(self):
144 self._struct_parser = StructParser(C_DEFINITION)
146 def extract(self, inpath: Path, outdir: Path) -> ExtractResult: # noqa: C901
147 fs = FileSystem(outdir)
148 with File.from_path(inpath) as file:
149 label = self._find_label(file)
150 label_offset = label.sector * SECTOR_SIZE
152 file.seek(label_offset + label.header_size, io.SEEK_SET)
153 pv_header = self._struct_parser.parse(
154 "lvm_pv_header_t", file, Endian.LITTLE
155 )
156 if pv_header.device_size > file.size():
157 raise InvalidInputFormat("LVM PV size exceeds the input size.")
159 label_end = label_offset + SECTOR_SIZE
160 data_areas = self._read_descriptors(file, label_end)
161 metadata_areas = self._read_descriptors(file, label_end)
162 if not data_areas:
163 raise InvalidInputFormat("LVM PV has no data area.")
164 if not metadata_areas:
165 raise InvalidInputFormat("LVM PV has no metadata area.")
167 data_area = data_areas[0]
168 self._validate_area(data_area, pv_header.device_size, "data")
169 metadata_area = metadata_areas[0]
170 self._validate_area(metadata_area, pv_header.device_size, "metadata")
171 mda_offset = metadata_area.area_offset
173 file.seek(mda_offset, io.SEEK_SET)
174 mda = self._struct_parser.parse(
175 "lvm_metadata_area_header_t", file, Endian.LITTLE
176 )
177 self._validate_mda(
178 file, mda, mda_offset, metadata_area, pv_header.device_size
179 )
181 locn = next(
182 (
183 location
184 for location in mda.locns
185 if location.data_offset
186 and location.data_size
187 and not location.flags & RAW_LOCATION_IGNORED
188 ),
189 None,
190 )
191 if locn is None:
192 raise InvalidInputFormat("LVM metadata area has no usable location.")
194 text = self._decode_metadata(file, mda_offset, mda.size, locn)
195 metadata = parse_lvm_metadata(text)
197 vg = self._get_vg(metadata)
198 try:
199 pv_name = self._get_pv_name(vg, pv_header.uuid.decode("ascii"))
200 except UnicodeDecodeError as e:
201 raise InvalidInputFormat(
202 "LVM physical volume UUID is not ASCII encoded"
203 ) from e
204 extent_size = vg.get("extent_size")
205 if not isinstance(extent_size, int) or extent_size <= 0:
206 raise InvalidInputFormat("LVM volume group has invalid extent size.")
207 extent_bytes = extent_size * SECTOR_SIZE
208 pe_start = data_area.area_offset
209 data_area_size = data_area.area_size or pv_header.device_size - pe_start
211 logical_volumes = vg.get("logical_volumes")
212 if not isinstance(logical_volumes, dict):
213 raise InvalidInputFormat("LVM volume group has invalid logical volumes")
215 for lv_name, lv in logical_volumes.items():
216 self._extract_lv(
217 file,
218 fs,
219 lv_name,
220 lv,
221 pv_name,
222 pe_start,
223 data_area_size,
224 extent_bytes,
225 )
227 return ExtractResult(reports=fs.problems)
229 def _find_label(self, file: File):
230 for sector in range(LABEL_SCAN_SECTORS):
231 label_offset = sector * SECTOR_SIZE
232 file.seek(label_offset, io.SEEK_SET)
233 if file.read(8) != b"LABELONE":
234 continue
236 file.seek(label_offset, io.SEEK_SET)
237 label = self._struct_parser.parse("lvm_label_header_t", file, Endian.LITTLE)
238 if (
239 label.sector == sector
240 and LABEL_HEADER_SIZE
241 <= label.header_size
242 <= SECTOR_SIZE - PV_HEADER_SIZE
243 and label.type == b"LVM2 001"
244 and self._valid_label_crc(file, label_offset, label.crc)
245 ):
246 return label
248 raise InvalidInputFormat("LVM label not found in the first four sectors.")
250 @staticmethod
251 def _valid_label_crc(file: File, label_offset: int, expected_crc: int) -> bool:
252 file.seek(label_offset + 20, io.SEEK_SET)
253 return _lvm_crc32(file.read(SECTOR_SIZE - 20)) == expected_crc
255 @staticmethod
256 def _read_metadata(file: File, mda_offset: int, mda_size: int, locn) -> bytes:
257 if (
258 mda_size < SECTOR_SIZE
259 or not SECTOR_SIZE <= locn.data_offset < mda_size
260 or locn.data_size > mda_size - SECTOR_SIZE
261 ):
262 raise InvalidInputFormat("Invalid LVM metadata location.")
264 first_size = min(locn.data_size, mda_size - locn.data_offset)
265 file.seek(mda_offset + locn.data_offset, io.SEEK_SET)
266 data = file.read(first_size)
268 remaining = locn.data_size - first_size
269 if remaining:
270 file.seek(mda_offset + SECTOR_SIZE, io.SEEK_SET)
271 data += file.read(remaining)
273 if len(data) != locn.data_size:
274 raise InvalidInputFormat("Truncated LVM metadata location.")
275 return data
277 @staticmethod
278 def _decode_metadata(file: File, mda_offset: int, mda_size: int, locn) -> str:
279 data = LVM2Extractor._read_metadata(file, mda_offset, mda_size, locn)
280 if _lvm_crc32(data) != locn.crc:
281 raise InvalidInputFormat("Invalid LVM metadata checksum.")
282 try:
283 return data.decode("utf-8")
284 except UnicodeDecodeError as exc:
285 raise InvalidInputFormat("LVM metadata is not valid UTF-8.") from exc
287 @staticmethod
288 def _validate_mda(file: File, mda, mda_offset: int, area, pv_size: int) -> None:
289 file.seek(mda_offset + 4, io.SEEK_SET)
290 if (
291 _lvm_crc32(file.read(SECTOR_SIZE - 4)) != mda.crc
292 or mda.magic != MDA_MAGIC
293 or mda.version != MDA_VERSION
294 or mda.offset != mda_offset
295 or mda.size < SECTOR_SIZE
296 or mda.size > pv_size - mda_offset
297 or (area.area_size and mda.size > area.area_size)
298 ):
299 raise InvalidInputFormat("Invalid LVM metadata area header.")
301 @staticmethod
302 def _validate_area(area, pv_size: int, area_name: str) -> None:
303 if area.area_offset >= pv_size or (
304 area.area_size and area.area_size > pv_size - area.area_offset
305 ):
306 raise InvalidInputFormat(f"Invalid LVM {area_name} area descriptor.")
308 @staticmethod
309 def _get_vg(metadata: dict) -> dict:
310 """Locate the VG block — the only top-level value that is a dict."""
311 for body in metadata.values():
312 if isinstance(body, dict):
313 return body
314 raise InvalidInputFormat("LVM metadata has no volume group block.")
316 @staticmethod
317 def _get_pv_name(vg: dict, pv_uuid: str) -> str:
318 """Match the binary PV UUID to the metadata's physical_volumes entry."""
319 for name, body in vg["physical_volumes"].items():
320 if body["id"].replace("-", "") == pv_uuid:
321 return name
322 raise InvalidInputFormat("PV UUID not found in volume group metadata.")
324 def _extract_lv(
325 self,
326 file: File,
327 fs: FileSystem,
328 lv_name: str,
329 lv: dict,
330 pv_name: str,
331 pe_start: int,
332 data_area_size: int,
333 extent_bytes: int,
334 ):
335 out_path = Path(f"{lv_name}.img")
336 with fs.open(out_path, "wb+") as outfile:
337 for key, seg in lv.items():
338 # filter to segment sub-sections; "segment_count" also matches the prefix but is an int
339 if not (key.startswith("segment") and isinstance(seg, dict)):
340 continue
341 self._extract_segment(
342 file,
343 outfile,
344 fs,
345 lv_name,
346 key,
347 seg,
348 pv_name,
349 pe_start,
350 data_area_size,
351 extent_bytes,
352 )
354 @staticmethod
355 def _extract_segment(
356 file: File,
357 outfile: BinaryIO,
358 fs: FileSystem,
359 lv_name: str,
360 key: str,
361 seg: dict,
362 pv_name: str,
363 pe_start: int,
364 data_area_size: int,
365 extent_bytes: int,
366 ) -> None:
367 resolution = "Segment skipped, output file will have a gap."
368 if seg.get("type") not in {"linear", "striped"} or seg.get("stripe_count") != 1:
369 fs.record_problem(
370 ExtractionProblem(
371 problem=f"{lv_name}/{key}: unsupported segment (type={seg.get('type')!r})",
372 resolution=resolution,
373 )
374 )
375 return
377 stripes = seg.get("stripes")
378 if not (
379 isinstance(stripes, list)
380 and len(stripes) == 2
381 and isinstance(stripes[0], str)
382 and isinstance(stripes[1], int)
383 and stripes[1] >= 0
384 ):
385 fs.record_problem(
386 ExtractionProblem(
387 problem=f"{lv_name}/{key}: invalid stripes list",
388 resolution=resolution,
389 )
390 )
391 return
392 if stripes[0] != pv_name:
393 fs.record_problem(
394 ExtractionProblem(
395 problem=f"{lv_name}/{key}: segment lives on foreign PV {stripes[0]!r}",
396 resolution=resolution,
397 )
398 )
399 return
401 start_extent = seg.get("start_extent")
402 extent_count = seg.get("extent_count")
403 if not (
404 isinstance(start_extent, int)
405 and start_extent >= 0
406 and isinstance(extent_count, int)
407 and extent_count > 0
408 ):
409 fs.record_problem(
410 ExtractionProblem(
411 problem=f"{lv_name}/{key}: invalid extent range",
412 resolution=resolution,
413 )
414 )
415 return
417 pe_index = stripes[1]
418 src = pe_start + pe_index * extent_bytes
419 dst = start_extent * extent_bytes
420 length = extent_count * extent_bytes
421 relative_end = pe_index * extent_bytes + length
422 if relative_end > data_area_size:
423 fs.record_problem(
424 ExtractionProblem(
425 problem=f"{lv_name}/{key}: segment exceeds the local PV data area",
426 resolution=resolution,
427 )
428 )
429 return
431 outfile.seek(dst, io.SEEK_SET)
432 for chunk in iterate_file(file, src, length):
433 outfile.write(chunk)
435 def _read_descriptors(self, file: File, end_offset: int) -> list:
436 descs = []
437 while True:
438 if file.tell() + DESCRIPTOR_SIZE > end_offset:
439 raise InvalidInputFormat("Unterminated LVM area descriptor list.")
440 d = self._struct_parser.parse(
441 "lvm_data_area_descriptor_t", file, Endian.LITTLE
442 )
443 if d.area_offset == 0 and d.area_size == 0:
444 return descs
445 descs.append(d)
448class LVM2Handler(StructHandler):
449 NAME = "lvm2"
451 PATTERNS = [
452 HexString("""
453 4c 41 42 45 4c 4f 4e 45 // LABELONE
454 [16] // sector(8) + crc(4) + data_offset(4)
455 4c 56 4d 32 20 30 30 31 // LVM2 001
456 """),
457 ]
458 EXTRACTOR = LVM2Extractor()
459 C_DEFINITIONS = C_DEFINITION
461 HEADER_STRUCT = "lvm_label_header_t"
463 DOC = HandlerDoc(
464 name="LVM2",
465 description="LVM2 (Logical Volume Manager 2) is a volume management system for Linux block storage, grouping physical volumes (PVs) into volume groups (VGs) that expose logical volumes (LVs) as resizable virtual block devices. Each PV carries text-format metadata describing the VG layout and a data area holding LV contents as fixed-size physical extents.",
466 handler_type=HandlerType.FILESYSTEM,
467 vendor=None,
468 references=[
469 Reference(
470 title="LVM2 on-disk format (libvslvm)",
471 url="https://github.com/libyal/libvslvm/blob/main/documentation/Logical%20Volume%20Manager%20(LVM)%20format.asciidoc",
472 ),
473 ],
474 limitations=[
475 "Multi-PV volume groups produce one partial LV image per PV chunk. The data is preserved across all extractions, but combining the partials into a single LV image is left to the user.",
476 "Only linear segments (striped with stripe_count=1) are supported. Other segment types (multi-stripe, mirror, raid, thin, snapshot, cache) require cross-PV reassembly or a separate format parser.",
477 ],
478 )
480 def is_valid_header(self, header, start_offset: int) -> bool:
481 return (
482 header.sector < LABEL_SCAN_SECTORS
483 and LABEL_HEADER_SIZE <= header.header_size <= SECTOR_SIZE - PV_HEADER_SIZE
484 and header.sector * SECTOR_SIZE <= start_offset
485 )
487 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
488 header = self.parse_header(file, Endian.LITTLE)
490 if not self.is_valid_header(header, start_offset):
491 raise InvalidInputFormat("Invalid LVM label header.")
493 pv_start = start_offset - header.sector * SECTOR_SIZE
495 file.seek(start_offset + 20, io.SEEK_SET)
496 if _lvm_crc32(file.read(SECTOR_SIZE - 20)) != header.crc:
497 raise InvalidInputFormat("Invalid LVM label checksum.")
499 file.seek(start_offset + header.header_size, io.SEEK_SET)
500 pv_header = self.cparser_le.lvm_pv_header_t(file)
502 pv_end = pv_start + pv_header.device_size
503 if pv_header.device_size == 0 or pv_end > file.size():
504 raise InvalidInputFormat("Invalid LVM PV device size.")
506 return ValidChunk(
507 start_offset=pv_start,
508 end_offset=pv_end,
509 )