Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/arpy.py: 55%

1# 

2# Copyright 2011 Stanisław Pitucha. All rights reserved. 

3# Copyright 2013 Helmut Grohne. All rights reserved. 

4# 

5# Redistribution and use in source and binary forms, with or without modification, are 

6# permitted provided that the following conditions are met: 

7# 

8# 1. Redistributions of source code must retain the above copyright notice, this list of 

9# conditions and the following disclaimer. 

10# 

11# 2. Redistributions in binary form must reproduce the above copyright notice, this list 

12# of conditions and the following disclaimer in the documentation and/or other materials 

13# provided with the distribution. 

14# 

15# THIS SOFTWARE IS PROVIDED BY Stanisław Pitucha ``AS IS'' AND ANY EXPRESS OR IMPLIED 

16# WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND 

17# FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL Stanisław Pitucha OR 

18# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 

19# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 

20# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON 

21# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 

22# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF 

23# ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 

24# 

25# The views and conclusions contained in the software and documentation are those of the 

26# authors and should not be interpreted as representing official policies, either expressed 

27# or implied, of Stanisław Pitucha. 

28# 

29 

30 

31"""arpy module can be used for reading `ar` files' headers, as well as 

32accessing the data contained in the archive. Archived files are accessible via 

33file-like objects. 

34 

35Support for both GNU and BSD extended length filenames is included. 

36 

37In order to read the file, create a new proxy with: 

38ar = arpy.Archive('some_ar_file') 

39ar.read_all_headers() 

40 

41The list of file names can be listed through: 

42ar.archived_files.keys() 

43 

44Files themselves can be opened by getting the value of: 

45f = ar.archived_files[b'filename'] 

46 

47and read through: 

48f.read([length]) 

49 

50random access through seek and tell functions is supported on the archived files. 

51 

52zipfile-like interface is also available: 

53 

54ar.namelist() will return a list of names (with possible duplicates) 

55ar.infolist() will return a list of headers 

56 

57Use ar.open(name / header) to get the specific file. 

58 

59You can also use context manager syntax with either the ar file or its contents. 

60""" # noqa: D205 

61 

62import io 

63import struct 

64from pathlib import Path 

65from typing import BinaryIO, cast 

66 

67HEADER_BSD = 1 

68HEADER_GNU = 2 

69HEADER_GNU_TABLE = 3 

70HEADER_GNU_SYMBOLS = 4 

71HEADER_NORMAL = 5 

72HEADER_TYPES = { 

73 HEADER_BSD: "BSD", 

74 HEADER_GNU: "GNU", 

75 HEADER_GNU_TABLE: "GNU_TABLE", 

76 HEADER_GNU_SYMBOLS: "GNU_SYMBOLS", 

77 HEADER_NORMAL: "NORMAL", 

78} 

79 

80GLOBAL_HEADER_LEN = 8 

81HEADER_LEN = 60 

82 

83 

84class ArchiveFormatError(Exception): 

85 """Raised on problems with parsing the archive headers.""" 

86 

87 

88class ArchiveAccessError(IOError): 

89 """Raised on problems with accessing the archived files.""" 

90 

91 

92class ArchiveFileHeader: 

93 """File header of an archived file, or a special data segment.""" 

94 

95 def __init__(self, header: bytes, offset: int) -> None: # noqa: C901 

96 """Create a new header from binary data starting at a specified offset.""" 

97 name, timestamp, uid, gid, mode, size, magic = struct.unpack( 

98 "16s 12s 6s 6s 8s 10s 2s", header 

99 ) 

100 if magic != b"\x60\x0a": 

101 raise ArchiveFormatError("file header magic doesn't match") 

102 

103 if name.startswith(b"#1/"): 

104 self.type = HEADER_BSD 

105 elif name.startswith((b"//", b"/SYM64/")): 

106 self.type = HEADER_GNU_TABLE 

107 elif name.strip() == b"/": 

108 self.type = HEADER_GNU_SYMBOLS 

109 elif name.startswith(b"/"): 

110 self.type = HEADER_GNU 

111 else: 

112 self.type = HEADER_NORMAL 

113 

114 try: 

115 self.size = int(size) 

116 

117 if self.type in (HEADER_NORMAL, HEADER_BSD, HEADER_GNU): 

118 self.timestamp = int(timestamp) 

119 if uid.strip(): 

120 self.uid = cast(int | None, int(uid)) 

121 else: 

122 self.uid = None 

123 if gid.strip(): 

124 self.gid = cast(int | None, int(gid)) 

125 else: 

126 self.gid = None 

127 self.mode = int(mode, 8) 

128 

129 except ValueError as err: 

130 raise ArchiveFormatError( 

131 "cannot convert file header fields to integers" 

132 ) from err 

133 

134 self.offset = offset 

135 name = name.rstrip() 

136 if len(name) > 1: 

137 name = name.rstrip(b"/") 

138 

139 if self.type == HEADER_NORMAL: 

140 self.name = name 

141 self.file_offset = cast(int | None, offset + HEADER_LEN) 

142 else: 

143 self.name = None 

144 self.proxy_name = name 

145 self.file_offset = None 

146 

147 def __repr__(self) -> str: 

148 """Create a human-readable summary of a header.""" 

149 return f"""<ArchiveFileHeader: "{self.name}" type:{HEADER_TYPES[self.type]} size:{self.size}>""" 

150 

151 

152class ArchiveFileData(io.IOBase): 

153 """File-like object used for reading an archived file.""" 

154 

155 def __init__(self, ar_obj: "Archive", header: ArchiveFileHeader) -> None: 

156 """Create a new proxy for the archived file, reusing the archive's file descriptor.""" 

157 self.header = header 

158 self.arobj = ar_obj 

159 self.last_offset = 0 

160 

161 def read(self, size: int | None = None) -> bytes: 

162 """Read the data from the archived file, simulates file.read.""" 

163 if size is None: 

164 size = self.header.size 

165 

166 if self.header.size < self.last_offset + size: 

167 size = self.header.size - self.last_offset 

168 

169 self.arobj.seek(cast(int, self.header.file_offset) + self.last_offset) 

170 data = self.arobj.read(size) 

171 if len(data) < size: 

172 raise ArchiveAccessError("incorrect archive file") 

173 

174 self.last_offset += size 

175 return data 

176 

177 def tell(self) -> int: 

178 """Return the position in archived file, simulates file.tell.""" 

179 return self.last_offset 

180 

181 def seek(self, offset: int, whence: int = 0) -> int: 

182 """Set the position in archived file, simulates file.seek.""" 

183 if whence == 0: 

184 pass # absolute 

185 elif whence == 1: 

186 offset += self.last_offset 

187 elif whence == 2: 

188 offset += self.header.size 

189 else: 

190 raise ArchiveAccessError("invalid argument") 

191 

192 if offset < 0 or offset > self.header.size: 

193 raise ArchiveAccessError("incorrect file position") 

194 self.last_offset = offset 

195 

196 return offset 

197 

198 def seekable(self) -> bool: 

199 return self.arobj.seekable 

200 

201 def __enter__(self) -> "ArchiveFileData": 

202 return self 

203 

204 def __exit__(self, _exc_type, _exc_value, _traceback): 

205 return 

206 

207 

208class ArchiveFileDataThin(ArchiveFileData): 

209 """File-like object used for reading a thin archived file.""" 

210 

211 def __init__(self, ar_obj: "Archive", header: ArchiveFileHeader) -> None: 

212 ArchiveFileData.__init__(self, ar_obj, header) 

213 if header.name is not None: 

214 self.file_path = Path(ar_obj.file.name).parent / header.name.decode() 

215 

216 def read(self, size: int | None = None) -> bytes: 

217 """Read the data from the archived file, simulates file.read.""" 

218 if size is None: 

219 size = self.header.size - self.last_offset 

220 

221 with self.file_path.open("rb") as f: 

222 f.seek(self.last_offset) 

223 data = f.read(size) 

224 

225 if len(data) < size: 

226 raise ArchiveAccessError("incorrect archive file") 

227 self.last_offset += size 

228 return data 

229 

230 

231class Archive: 

232 """Archive object allowing reading of *.ar files.""" 

233 

234 def __init__( 

235 self, filename: str | None = None, fileobj: BinaryIO | None = None 

236 ) -> None: 

237 self.headers = cast(list[ArchiveFileHeader], []) 

238 if fileobj: 

239 self.file = fileobj 

240 elif filename: 

241 self.file = Path(filename).open("rb") # noqa: SIM115 

242 else: 

243 raise ValueError("either filename or fileobj argument needs to be given") 

244 self.position = 0 

245 self.reached_eof = False 

246 self._detect_seekable() 

247 global_header = self.read(GLOBAL_HEADER_LEN) 

248 if global_header == b"!<arch>\n": 

249 self.file_data_class = ArchiveFileData 

250 elif global_header == b"!<thin>\n": 

251 self.file_data_class = ArchiveFileDataThin 

252 else: 

253 raise ArchiveFormatError("file is missing the global header") 

254 

255 self.next_header_offset = GLOBAL_HEADER_LEN 

256 self.gnu_table_data: bytes | None = None 

257 self.gnu_table_separator = b"\n" 

258 self.gnu_name_cache: dict[int, bytes] = {} 

259 self.archived_files = cast(dict[bytes, ArchiveFileData], {}) 

260 

261 def _detect_seekable(self) -> None: 

262 if hasattr(self.file, "seekable"): 

263 self.seekable = self.file.seekable() 

264 else: 

265 try: 

266 # .tell() will raise an exception as well 

267 self.file.tell() 

268 self.seekable = True 

269 except Exception: 

270 self.seekable = False 

271 

272 def read(self, length: int) -> bytes: 

273 data = self.file.read(length) 

274 self.position += len(data) 

275 return data 

276 

277 def seek(self, offset: int) -> None: 

278 if self.seekable: 

279 self.file.seek(offset) 

280 self.position = self.file.tell() 

281 elif offset < self.position: 

282 raise ArchiveAccessError( 

283 "cannot go back when reading archive from a stream" 

284 ) 

285 else: 

286 # emulate seek 

287 while self.position < offset: 

288 if not self.read(min(4096, offset - self.position)): 

289 # reached EOF before target offset 

290 self.reached_eof = True 

291 return 

292 

293 def __read_file_header(self, offset: int) -> ArchiveFileHeader | None: 

294 """Read and returns a single new file header.""" 

295 self.seek(offset) 

296 

297 header = self.read(HEADER_LEN) 

298 

299 if len(header) == 0: 

300 self.reached_eof = True 

301 return None 

302 if len(header) < HEADER_LEN: 

303 raise ArchiveFormatError("file header too short") 

304 

305 file_header = ArchiveFileHeader(header, offset) 

306 if file_header.type == HEADER_GNU_TABLE: 

307 self.__read_gnu_table(file_header.size) 

308 

309 add_len = self.__fix_name(file_header) 

310 file_header.file_offset = offset + HEADER_LEN + add_len 

311 

312 if offset == self.next_header_offset: 

313 new_offset = file_header.file_offset + file_header.size 

314 self.next_header_offset = Archive.__pad2(new_offset) 

315 

316 return file_header 

317 

318 def __read_gnu_table(self, size: int) -> None: 

319 """Read the table of filenames specific to GNU ar format.""" 

320 table_data = self.read(size) 

321 if len(table_data) != size: 

322 raise ArchiveFormatError("file too short to fit the names table") 

323 

324 self.gnu_table_data = table_data 

325 self.gnu_table_separator = b"\x00" if b"\x00" in table_data else b"\n" 

326 self.gnu_name_cache = {} 

327 

328 def __resolve_gnu_name(self, position: int) -> bytes: 

329 """Return the GNU extended filename starting at a table offset.""" 

330 cached_name = self.gnu_name_cache.get(position) 

331 if cached_name is not None: 

332 return cached_name 

333 

334 if self.gnu_table_data is None: 

335 raise ArchiveFormatError("file references a name not present in the index") 

336 

337 table_data = self.gnu_table_data 

338 separator = self.gnu_table_separator 

339 if position < 0 or position > len(table_data): 

340 raise ArchiveFormatError("file references a name not present in the index") 

341 if position and table_data[position - 1 : position] != separator: 

342 raise ArchiveFormatError("file references a name not present in the index") 

343 

344 end = table_data.find(separator, position) 

345 if end == -1: 

346 end = len(table_data) 

347 name = table_data[position:end].removesuffix(b"/") 

348 self.gnu_name_cache[position] = name 

349 return name 

350 

351 def __fix_name(self, header: ArchiveFileHeader) -> int: 

352 """Correct the long filename using the format-specific method. 

353 

354 That means either looking up the name in GNU filename table, or 

355 reading past the header in BSD ar files. 

356 """ 

357 if header.type == HEADER_NORMAL: 

358 pass 

359 

360 elif header.type == HEADER_BSD: 

361 filename_len = Archive.__get_bsd_filename_len(header.proxy_name) 

362 

363 # BSD format includes the filename in the file size 

364 header.size -= filename_len 

365 

366 self.seek(header.offset + HEADER_LEN) 

367 header.name = self.read(filename_len) 

368 return filename_len 

369 

370 elif header.type == HEADER_GNU_TABLE: 

371 header.name = "*GNU_TABLE*" 

372 

373 elif header.type == HEADER_GNU: 

374 gnu_position = int(header.proxy_name[1:]) 

375 header.name = self.__resolve_gnu_name(gnu_position) 

376 

377 elif header.type == HEADER_GNU_SYMBOLS: 

378 pass 

379 

380 return 0 

381 

382 @staticmethod 

383 def __pad2(num: int) -> int: 

384 """Return a 2-aligned offset.""" 

385 if num % 2 == 0: 

386 return num 

387 return num + 1 

388 

389 @staticmethod 

390 def __get_bsd_filename_len(name: bytes) -> int: 

391 """Return the length of the filename for a BSD style header.""" 

392 filename_len = name[3:] 

393 return int(filename_len) 

394 

395 def read_next_header(self) -> ArchiveFileHeader | None: 

396 """Read a single new header, returning a its representation, or None at the end of file.""" 

397 header = self.__read_file_header(self.next_header_offset) 

398 if header is not None: 

399 self.headers.append(header) 

400 if header.type in (HEADER_BSD, HEADER_NORMAL, HEADER_GNU): 

401 self.archived_files[header.name] = self.file_data_class(self, header) # pyright: ignore[reportArgumentType] 

402 

403 return header 

404 

405 def __next__(self) -> ArchiveFileData: 

406 while True: 

407 header = self.read_next_header() 

408 if header is None: 

409 raise StopIteration 

410 if header.type in (HEADER_BSD, HEADER_NORMAL, HEADER_GNU): 

411 return self.archived_files[header.name] # pyright: ignore[reportArgumentType] 

412 

413 next = __next__ 

414 

415 def __iter__(self) -> "Archive": 

416 return self 

417 

418 def read_all_headers(self) -> None: 

419 """Read all headers.""" 

420 if self.reached_eof: 

421 return 

422 

423 while self.read_next_header() is not None: 

424 pass 

425 

426 def close(self) -> None: 

427 """Close the archive file descriptor.""" 

428 self.file.close() 

429 

430 ### implement a zipfile-like interface as well 

431 

432 def namelist(self) -> list[bytes]: 

433 """Return the names of files stored in the archive. 

434 

435 If there are multiple files of the same name, there may be duplicates in the list. 

436 """ 

437 self.read_all_headers() 

438 return [ # pyright: ignore[reportReturnType] 

439 header.name 

440 for header in self.headers 

441 if header.type in (HEADER_BSD, HEADER_NORMAL, HEADER_GNU) 

442 ] 

443 

444 def infolist(self) -> list[ArchiveFileHeader]: 

445 """Return the headers of files stored in the archive. 

446 

447 These can be used with .open() to get the contents. 

448 """ 

449 self.read_all_headers() 

450 return [ 

451 header 

452 for header in self.headers 

453 if header.type in (HEADER_BSD, HEADER_NORMAL, HEADER_GNU) 

454 ] 

455 

456 def open(self, name: bytes | ArchiveFileHeader) -> ArchiveFileData: 

457 """Return a file-like object based on the provided name or header. 

458 

459 The name can be either a filename, or a header obtained from .read_next_header() or .infolist() 

460 """ 

461 self.read_all_headers() 

462 

463 if isinstance(name, bytes): 

464 ar_file = self.archived_files.get(name) 

465 if ar_file is None: 

466 raise KeyError(f"There is no item named {name!r} in the archive") 

467 

468 return ar_file 

469 

470 if isinstance(name, ArchiveFileHeader): 

471 if name not in self.headers: 

472 raise KeyError("Provided header does not match this archive") 

473 

474 return ArchiveFileData(ar_obj=self, header=name) 

475 

476 raise ValueError( 

477 f"Can't look up file using type {type(name)}, expected bytes or ArchiveFileHeader" 

478 ) 

479 

480 def __enter__(self) -> "Archive": 

481 return self 

482 

483 def __exit__(self, _exc_type, _exc_value, _traceback): 

484 self.close() 

485 return False 

486 

487 

488if __name__ == "__main__": 

489 import sys 

490 

491 ar = Archive(sys.argv[1]) 

492 ar.read_all_headers() 

493 

494 for key in ar.archived_files: 

495 print(key) # noqa: T201