Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/midx.py: 16%

1# midx.py -- Multi-Pack-Index (MIDX) support

4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later

5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU

6# General Public License as published by the Free Software Foundation; version 2.0

7# or (at your option) any later version. You can redistribute it and/or

8# modify it under the terms of either of these two licenses.

10# Unless required by applicable law or agreed to in writing, software

11# distributed under the License is distributed on an "AS IS" BASIS,

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

13# See the License for the specific language governing permissions and

14# limitations under the License.

15#

16# You should have received a copy of the licenses; if not, see

17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License

18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache

19# License, Version 2.0.

20#

22"""Multi-Pack-Index (MIDX) support.

24A multi-pack-index (MIDX) provides a single index that covers multiple pack files,

25enabling fast object lookup across all packs without opening each pack index.

27The MIDX file format consists of:

28- A header with signature, version, and hash algorithm

29- A chunk lookup table

30- Multiple chunks containing pack names, OID fanout, OID lookup, and object offsets

31- A trailer with checksum

33This module provides:

34- Reading MIDX files

35- Writing MIDX files

36- Integration with pack-based object stores

38Limitations:

39- Incremental MIDX chains are not yet supported (base_midx_files must be 0)

40- BTMP (bitmapped packfiles) chunk is not yet implemented

41- RIDX (reverse index) chunk is not yet implemented

43Note: Incremental MIDX chains were introduced in Git 2.47 as an experimental

44feature, where multiple MIDX files can be chained together. The format includes

45a base_midx_files field in the header and uses a multi-pack-index.d/ directory

46with a multi-pack-index-chain file. This feature is not yet supported by Dulwich

47as the specification is still evolving.

48"""

50__all__ = [

51 "CHUNK_BTMP",

52 "CHUNK_LOFF",

53 "CHUNK_OIDF",

54 "CHUNK_OIDL",

55 "CHUNK_OOFF",

56 "CHUNK_PNAM",

57 "CHUNK_RIDX",

58 "HASH_ALGORITHM_SHA1",

59 "HASH_ALGORITHM_SHA256",

60 "MIDX_SIGNATURE",

61 "MIDX_VERSION",

62 "MultiPackIndex",

63 "has_mmap",

64 "load_midx",

65 "load_midx_file",

66 "write_midx",

67 "write_midx_file",

68]

70import mmap

71import os

72import struct

73from collections.abc import Iterator

74from io import UnsupportedOperation

75from typing import IO, Any

77from .file import GitFile, _GitFile

78from .objects import ObjectID, RawObjectID

79from .pack import SHA1Writer

81has_mmap = True

83# MIDX signature

84MIDX_SIGNATURE = b"MIDX"

86# MIDX version

87MIDX_VERSION = 1

89# Chunk identifiers (4 bytes each)

90CHUNK_PNAM = b"PNAM" # Packfile names

91CHUNK_OIDF = b"OIDF" # OID fanout table

92CHUNK_OIDL = b"OIDL" # OID lookup table

93CHUNK_OOFF = b"OOFF" # Object offsets

94CHUNK_LOFF = b"LOFF" # Large offsets (optional)

95CHUNK_BTMP = b"BTMP" # Bitmapped packfiles (optional)

96CHUNK_RIDX = b"RIDX" # Reverse index (optional)

98# Hash algorithm identifiers

99HASH_ALGORITHM_SHA1 = 1

100HASH_ALGORITHM_SHA256 = 2

101

102

103class MultiPackIndex:

104 """Multi-pack-index for efficient object lookup across multiple pack files."""

105

106 def __init__(

107 self,

108 filename: str | os.PathLike[str],

109 file: IO[bytes] | _GitFile | None = None,

110 contents: bytes | None = None,

111 size: int | None = None,

112 ) -> None:

113 """Initialize a MultiPackIndex.

114

115 Args:

116 filename: Path to the MIDX file

117 file: Optional file object

118 contents: Optional mmap'd contents

119 size: Optional size of the MIDX file

120 """

121 self._filename = os.fspath(filename)

122 self._file = file

123 self._size = size

124

125 # Instance variables that will be set during parsing

126 self.version: int

127 self.hash_algorithm: int

128 self.hash_size: int

129 self.chunk_count: int

130 self.base_midx_files: int

131 self.pack_count: int

132 self.pack_names: list[str]

133 self.object_count: int

134 self._contents: bytes | Any

135 self._chunks: dict[bytes, int]

136 self._fanout_table: list[int]

137 self._oidl_offset: int

138 self._ooff_offset: int

139 self._loff_offset: int

140

141 # Load file contents

142 if contents is None:

143 if file is None:

144 with GitFile(filename, "rb") as f:

145 self._contents, self._size = self._load_file_contents(f, size)

146 else:

147 self._contents, self._size = self._load_file_contents(file, size)

148 else:

149 self._contents = contents

150

151 # Parse header

152 self._parse_header()

153

154 # Parse chunk lookup table

155 self._parse_chunk_table()

156

157 def _load_file_contents(

158 self, f: IO[bytes] | _GitFile, size: int | None = None

159 ) -> tuple[bytes | Any, int]:

160 """Load contents from a file, preferring mmap when possible.

161

162 Args:

163 f: File-like object to load

164 size: Expected size, or None to determine from file

165

166 Returns:

167 Tuple of (contents, size)

168 """

169 try:

170 fd = f.fileno()

171 except (UnsupportedOperation, AttributeError):

172 fd = None

173

174 # Attempt to use mmap if possible

175 if fd is not None:

176 if size is None:

177 size = os.fstat(fd).st_size

178 if has_mmap:

179 try:

180 contents = mmap.mmap(fd, size, access=mmap.ACCESS_READ)

181 except (OSError, ValueError):

182 # Can't mmap - perhaps a socket or invalid file descriptor

183 pass

184 else:

185 return contents, size

186

187 # Fall back to reading entire file into memory

188 contents_bytes = f.read()

189 size = len(contents_bytes)

190 return contents_bytes, size

191

192 def _parse_header(self) -> None:

193 """Parse the MIDX header."""

194 if len(self._contents) < 12:

195 raise ValueError("MIDX file too small")

196

197 # Check signature

198 signature = self._contents[0:4]

199 if signature != MIDX_SIGNATURE:

200 raise ValueError(f"Invalid MIDX signature: {signature!r}")

201

202 # Read version

203 self.version = self._contents[4]

204 if self.version != MIDX_VERSION:

205 raise ValueError(f"Unsupported MIDX version: {self.version}")

206

207 # Read object ID version (hash algorithm)

208 self.hash_algorithm = self._contents[5]

209 if self.hash_algorithm == HASH_ALGORITHM_SHA1:

210 self.hash_size = 20

211 elif self.hash_algorithm == HASH_ALGORITHM_SHA256:

212 self.hash_size = 32

213 else:

214 raise ValueError(f"Unknown hash algorithm: {self.hash_algorithm}")

215

216 # Read chunk count

217 self.chunk_count = self._contents[6]

218

219 # Read base MIDX files count (currently always 0)

220 self.base_midx_files = self._contents[7]

221 if self.base_midx_files != 0:

222 raise ValueError("Incremental MIDX not yet supported")

223

224 # Read pack file count

225 (self.pack_count,) = struct.unpack(">L", self._contents[8:12])

226

227 def _parse_chunk_table(self) -> None:

228 """Parse the chunk lookup table."""

229 self._chunks = {}

230

231 # Chunk table starts at offset 12

232 offset = 12

233

234 # Each chunk entry is 12 bytes (4-byte ID + 8-byte offset)

235 for i in range(self.chunk_count + 1): # +1 for terminator

236 chunk_id = self._contents[offset : offset + 4]

237 (chunk_offset,) = struct.unpack(

238 ">Q", self._contents[offset + 4 : offset + 12]

239 )

240

241 if chunk_id == b"\x00\x00\x00\x00":

242 # Terminator entry

243 break

244

245 self._chunks[chunk_id] = chunk_offset

246 offset += 12

247

248 # Parse required chunks

249 self._parse_pnam_chunk()

250 self._parse_oidf_chunk()

251 self._parse_oidl_chunk()

252 self._parse_ooff_chunk()

253

254 # Parse optional chunks

255 if CHUNK_LOFF in self._chunks:

256 self._parse_loff_chunk()

257

258 def _parse_pnam_chunk(self) -> None:

259 """Parse the Packfile Names (PNAM) chunk."""

260 if CHUNK_PNAM not in self._chunks:

261 raise ValueError("Required PNAM chunk not found")

262

263 offset = self._chunks[CHUNK_PNAM]

264 self.pack_names = []

265

266 # Find the end of the PNAM chunk (next chunk or end of chunks section)

267 next_offset = min(

268 (o for o in self._chunks.values() if o > offset),

269 default=len(self._contents),

270 )

271

272 # Parse null-terminated pack names

273 current = offset

274 while current < next_offset:

275 # Find the next null terminator

276 null_pos = self._contents.find(b"\x00", current, next_offset)

277 if null_pos == -1:

278 break

279

280 pack_name = self._contents[current:null_pos].decode("utf-8")

281 if pack_name: # Skip empty strings (padding)

282 self.pack_names.append(pack_name)

283 current = null_pos + 1

284

285 def _parse_oidf_chunk(self) -> None:

286 """Parse the OID Fanout (OIDF) chunk."""

287 if CHUNK_OIDF not in self._chunks:

288 raise ValueError("Required OIDF chunk not found")

289

290 offset = self._chunks[CHUNK_OIDF]

291 self._fanout_table = []

292

293 # Read 256 4-byte entries

294 for i in range(256):

295 (count,) = struct.unpack(

296 ">L", self._contents[offset + i * 4 : offset + i * 4 + 4]

297 )

298 self._fanout_table.append(count)

299

300 # Total object count is the last entry

301 self.object_count = self._fanout_table[255]

302

303 def _parse_oidl_chunk(self) -> None:

304 """Parse the OID Lookup (OIDL) chunk."""

305 if CHUNK_OIDL not in self._chunks:

306 raise ValueError("Required OIDL chunk not found")

307

308 self._oidl_offset = self._chunks[CHUNK_OIDL]

309

310 def _parse_ooff_chunk(self) -> None:

311 """Parse the Object Offsets (OOFF) chunk."""

312 if CHUNK_OOFF not in self._chunks:

313 raise ValueError("Required OOFF chunk not found")

314

315 self._ooff_offset = self._chunks[CHUNK_OOFF]

316

317 def _parse_loff_chunk(self) -> None:

318 """Parse the Large Offsets (LOFF) chunk."""

319 self._loff_offset = self._chunks[CHUNK_LOFF]

320

321 def __len__(self) -> int:

322 """Return the number of objects in this MIDX."""

323 return self.object_count

324

325 def _get_oid(self, index: int) -> RawObjectID:

326 """Get the object ID at the given index.

327

328 Args:

329 index: Index of the object

330

331 Returns:

332 Binary object ID

333 """

334 if index < 0 or index >= self.object_count:

335 raise IndexError(f"Index {index} out of range")

336

337 offset = self._oidl_offset + index * self.hash_size

338 return RawObjectID(self._contents[offset : offset + self.hash_size])

339

340 def _get_pack_info(self, index: int) -> tuple[int, int]:

341 """Get pack ID and offset for object at the given index.

342

343 Args:

344 index: Index of the object

345

346 Returns:

347 Tuple of (pack_id, offset)

348 """

349 if index < 0 or index >= self.object_count:

350 raise IndexError(f"Index {index} out of range")

351

352 # Each entry is 8 bytes (4-byte pack ID + 4-byte offset)

353 offset = self._ooff_offset + index * 8

354

355 (pack_id,) = struct.unpack(">L", self._contents[offset : offset + 4])

356 (pack_offset,) = struct.unpack(">L", self._contents[offset + 4 : offset + 8])

357

358 # Check if this is a large offset (MSB set)

359 if pack_offset & 0x80000000:

360 # Look up in LOFF chunk

361 if CHUNK_LOFF not in self._chunks:

362 raise ValueError("Large offset found but no LOFF chunk")

363

364 large_index = pack_offset & 0x7FFFFFFF

365 large_offset_pos = self._loff_offset + large_index * 8

366 (pack_offset,) = struct.unpack(

367 ">Q", self._contents[large_offset_pos : large_offset_pos + 8]

368 )

369

370 return pack_id, pack_offset

371

372 def object_offset(self, sha: ObjectID | RawObjectID) -> tuple[str, int] | None:

373 """Return the pack name and offset for the given object.

374

375 Args:

376 sha: Binary SHA-1 or SHA-256 hash

377

378 Returns:

379 Tuple of (pack_name, offset) or None if not found

380 """

381 if len(sha) != self.hash_size:

382 raise ValueError(

383 f"SHA size mismatch: expected {self.hash_size}, got {len(sha)}"

384 )

385

386 # Use fanout table to narrow search range

387 first_byte = sha[0]

388 start_idx = 0 if first_byte == 0 else self._fanout_table[first_byte - 1]

389 end_idx = self._fanout_table[first_byte]

390

391 # Binary search within the range

392 while start_idx < end_idx:

393 mid = (start_idx + end_idx) // 2

394 mid_sha = self._get_oid(mid)

395

396 if mid_sha == sha:

397 # Found it!

398 pack_id, offset = self._get_pack_info(mid)

399 return self.pack_names[pack_id], offset

400 elif mid_sha < sha:

401 start_idx = mid + 1

402 else:

403 end_idx = mid

404

405 return None

406

407 def __contains__(self, sha: ObjectID | RawObjectID) -> bool:

408 """Check if the given object SHA is in this MIDX.

409

410 Args:

411 sha: Binary SHA hash

412

413 Returns:

414 True if the object is in this MIDX

415 """

416 return self.object_offset(sha) is not None

417

418 def iterentries(self) -> Iterator[tuple[RawObjectID, str, int]]:

419 """Iterate over all entries in this MIDX.

420

421 Yields:

422 Tuples of (sha, pack_name, offset)

423 """

424 for i in range(self.object_count):

425 sha = self._get_oid(i)

426 pack_id, offset = self._get_pack_info(i)

427 pack_name = self.pack_names[pack_id]

428 yield sha, pack_name, offset

429

430 def close(self) -> None:

431 """Close the MIDX file and release mmap resources."""

432 # Close mmap'd contents first if it's an mmap object

433 if self._contents is not None and has_mmap:

434 if isinstance(self._contents, mmap.mmap):

435 self._contents.close()

436 self._contents = None

437

438 # Close file handle

439 if self._file is not None:

440 self._file.close()

441 self._file = None

442

443

444def load_midx(path: str | os.PathLike[str]) -> MultiPackIndex:

445 """Load a multi-pack-index file by path.

446

447 Args:

448 path: Path to the MIDX file

449

450 Returns:

451 A MultiPackIndex loaded from the given path

452 """

453 with GitFile(path, "rb") as f:

454 return load_midx_file(path, f)

455

456

457def load_midx_file(

458 path: str | os.PathLike[str], f: IO[bytes] | _GitFile

459) -> MultiPackIndex:

460 """Load a multi-pack-index from a file-like object.

461

462 Args:

463 path: Path for the MIDX file

464 f: File-like object

465

466 Returns:

467 A MultiPackIndex loaded from the given file

468 """

469 return MultiPackIndex(path, file=f)

470

471

472def write_midx(

473 f: IO[bytes],

474 pack_index_entries: list[tuple[str, list[tuple[RawObjectID, int, int | None]]]],

475 hash_algorithm: int = HASH_ALGORITHM_SHA1,

476) -> bytes:

477 """Write a multi-pack-index file.

478

479 Args:

480 f: File-like object to write to

481 pack_index_entries: List of (pack_name, entries) tuples where entries are

482 (sha, offset, crc32) tuples, sorted by SHA

483 hash_algorithm: Hash algorithm to use (1=SHA-1, 2=SHA-256)

484

485 Returns:

486 SHA-1 checksum of the written MIDX file

487 """

488 if hash_algorithm == HASH_ALGORITHM_SHA1:

489 hash_size = 20

490 elif hash_algorithm == HASH_ALGORITHM_SHA256:

491 hash_size = 32

492 else:

493 raise ValueError(f"Unknown hash algorithm: {hash_algorithm}")

494

495 # Wrap file in SHA1Writer to compute checksum

496 writer = SHA1Writer(f)

497

498 # Sort pack entries by pack name (required by Git)

499 pack_index_entries_sorted = sorted(pack_index_entries, key=lambda x: x[0])

500

501 # Collect all objects from all packs, deduplicating objects that appear

502 # in multiple packs. Git's MIDX format requires strictly increasing OIDs

503 # in the OIDL chunk, so duplicates must be resolved. When the same object

504 # is in multiple packs, we keep the occurrence from the first pack (in

505 # pack-name order), matching how Git itself handles duplicates.

506 seen: dict[RawObjectID, tuple[int, int]] = {} # sha -> (pack_id, offset)

507 pack_names: list[str] = []

508

509 for pack_id, (pack_name, entries) in enumerate(pack_index_entries_sorted):

510 pack_names.append(pack_name)

511 for sha, offset, _crc32 in entries:

512 if sha not in seen:

513 seen[sha] = (pack_id, offset)

514

515 all_objects: list[tuple[RawObjectID, int, int]] = [

516 (sha, pack_id, offset) for sha, (pack_id, offset) in seen.items()

517 ]

518

519 # Sort all objects by SHA

520 all_objects.sort(key=lambda x: x[0])

521

522 # Calculate offsets for chunks

523 num_packs = len(pack_names)

524 num_objects = len(all_objects)

525

526 # Header: 12 bytes

527 header_size = 12

528

529 # Chunk count: PNAM, OIDF, OIDL, OOFF, and optionally LOFF

530 # We'll determine if LOFF is needed later

531 chunk_count = 4 # PNAM, OIDF, OIDL, OOFF

532

533 # Check if we need LOFF chunk (for offsets >= 2^31)

534 need_loff = any(offset >= 2**31 for _sha, _pack_id, offset in all_objects)

535 if need_loff:

536 chunk_count += 1

537

538 # Chunk table: (chunk_count + 1) * 12 bytes (including terminator)

539 chunk_table_size = (chunk_count + 1) * 12

540

541 # Calculate chunk offsets

542 current_offset = header_size + chunk_table_size

543

544 # PNAM chunk: pack names as null-terminated strings, padded to 4-byte boundary

545 pnam_data = b"".join(name.encode("utf-8") + b"\x00" for name in pack_names)

546 # Pad to 4-byte boundary

547 pnam_padding = (4 - len(pnam_data) % 4) % 4

548 pnam_data += b"\x00" * pnam_padding

549 pnam_offset = current_offset

550 current_offset += len(pnam_data)

551

552 # OIDF chunk: 256 * 4 bytes

553 oidf_offset = current_offset

554 oidf_size = 256 * 4

555 current_offset += oidf_size

556

557 # OIDL chunk: num_objects * hash_size bytes

558 oidl_offset = current_offset

559 oidl_size = num_objects * hash_size

560 current_offset += oidl_size

561

562 # OOFF chunk: num_objects * 8 bytes (4 for pack_id + 4 for offset)

563 ooff_offset = current_offset

564 ooff_size = num_objects * 8

565 current_offset += ooff_size

566

567 # LOFF chunk (if needed): variable size

568 # We'll calculate the exact size when we know how many large offsets we have

569 loff_offset = current_offset if need_loff else 0

570 large_offsets: list[int] = []

571

572 # Calculate trailer offset (where checksum starts)

573 # We need to pre-calculate large offset count for accurate trailer offset

574 if need_loff:

575 # Count large offsets

576 large_offset_count = sum(1 for _, _, offset in all_objects if offset >= 2**31)

577 loff_size = large_offset_count * 8

578 trailer_offset = current_offset + loff_size

579 else:

580 trailer_offset = current_offset

581

582 # Write header

583 writer.write(MIDX_SIGNATURE) # 4 bytes: signature

584 writer.write(bytes([MIDX_VERSION])) # 1 byte: version

585 writer.write(bytes([hash_algorithm])) # 1 byte: hash algorithm

586 writer.write(bytes([chunk_count])) # 1 byte: chunk count

587 writer.write(bytes([0])) # 1 byte: base MIDX files (always 0)

588 writer.write(struct.pack(">L", num_packs)) # 4 bytes: pack count

589

590 # Write chunk table

591 chunk_table = [

592 (CHUNK_PNAM, pnam_offset),

593 (CHUNK_OIDF, oidf_offset),

594 (CHUNK_OIDL, oidl_offset),

595 (CHUNK_OOFF, ooff_offset),

596 ]

597 if need_loff:

598 chunk_table.append((CHUNK_LOFF, loff_offset))

599

600 for chunk_id, chunk_offset in chunk_table:

601 writer.write(chunk_id) # 4 bytes

602 writer.write(struct.pack(">Q", chunk_offset)) # 8 bytes

603

604 # Write terminator (points to where trailer/checksum starts)

605 writer.write(b"\x00\x00\x00\x00") # 4 bytes

606 writer.write(struct.pack(">Q", trailer_offset)) # 8 bytes

607

608 # Write PNAM chunk

609 writer.write(pnam_data)

610

611 # Write OIDF chunk (fanout table)

612 fanout: list[int] = [0] * 256

613 for sha, _pack_id, _offset in all_objects:

614 first_byte = sha[0]

615 fanout[first_byte] += 1

616

617 # Convert counts to cumulative

618 cumulative = 0

619 for i in range(256):

620 cumulative += fanout[i]

621 writer.write(struct.pack(">L", cumulative))

622

623 # Write OIDL chunk (object IDs)

624 for sha, _pack_id, _offset in all_objects:

625 writer.write(sha)

626

627 # Write OOFF chunk (pack ID and offset for each object)

628 for _sha, pack_id, offset in all_objects:

629 writer.write(struct.pack(">L", pack_id))

630

631 if offset >= 2**31:

632 # Use large offset table

633 large_offset_index = len(large_offsets)

634 large_offsets.append(offset)

635 # Set MSB to indicate large offset

636 writer.write(struct.pack(">L", 0x80000000 | large_offset_index))

637 else:

638 writer.write(struct.pack(">L", offset))

639

640 # Write LOFF chunk if needed

641 if need_loff:

642 for large_offset in large_offsets:

643 writer.write(struct.pack(">Q", large_offset))

644

645 # Write checksum

646 return writer.write_sha()

647

648

649def write_midx_file(

650 path: str | os.PathLike[str],

651 pack_index_entries: list[tuple[str, list[tuple[RawObjectID, int, int | None]]]],

652 hash_algorithm: int = HASH_ALGORITHM_SHA1,

653) -> bytes:

654 """Write a multi-pack-index file to disk.

655

656 Args:

657 path: Path where to write the MIDX file

658 pack_index_entries: List of (pack_name, entries) tuples where entries are

659 (sha, offset, crc32) tuples, sorted by SHA

660 hash_algorithm: Hash algorithm to use (1=SHA-1, 2=SHA-256)

661

662 Returns:

663 SHA-1 checksum of the written MIDX file

664 """

665 with GitFile(path, "wb") as f:

666 return write_midx(f, pack_index_entries, hash_algorithm)

667

668

669# TODO: Add support for incremental MIDX chains

670# TODO: Add support for BTMP and RIDX chunks for bitmap integration