cpython/Lib/zipfile/__init__.py at main · python/cpython · GitHub

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

"""

Read and write ZIP files.

XXX references to utf-8 need further investigation.

"""

import binascii

import io

import os

import shutil

import stat

import struct

import sys

import threading

import time

lazy import warnings

try:

import zlib # We may need its compression method

crc32 = zlib.crc32

except ImportError:

zlib = None

crc32 = binascii.crc32

try:

import bz2 # We may need its compression method

except ImportError:

bz2 = None

try:

import lzma # We may need its compression method

except ImportError:

lzma = None

try:

from compression import zstd # We may need its compression method

except ImportError:

zstd = None

__all__ = ["BadZipFile", "BadZipfile", "error",

"ZIP_STORED", "ZIP_DEFLATED", "ZIP_BZIP2", "ZIP_LZMA",

"ZIP_ZSTANDARD", "is_zipfile", "ZipInfo", "ZipFile", "PyZipFile",

"LargeZipFile", "Path"]

class BadZipFile(Exception):

pass

class LargeZipFile(Exception):

"""

Raised when writing a zipfile, the zipfile requires ZIP64 extensions

and those extensions are disabled.

"""

error = BadZipfile = BadZipFile # Pre-3.2 compatibility names

ZIP64_LIMIT = (1 << 31) - 1

ZIP_FILECOUNT_LIMIT = (1 << 16) - 1

ZIP_MAX_COMMENT = (1 << 16) - 1

# constants for Zip file compression methods

ZIP_STORED = 0

ZIP_DEFLATED = 8

ZIP_BZIP2 = 12

ZIP_LZMA = 14

ZIP_ZSTANDARD = 93

# Other ZIP compression methods not supported

DEFAULT_VERSION = 20

ZIP64_VERSION = 45

BZIP2_VERSION = 46

LZMA_VERSION = 63

ZSTANDARD_VERSION = 63

# we recognize (but not necessarily support) all features up to that version

MAX_EXTRACT_VERSION = 63

# Below are some formats and associated data for reading/writing headers using

# the struct module. The names and structures of headers/records are those used

# in the PKWARE description of the ZIP file format:

# http://www.pkware.com/documents/casestudies/APPNOTE.TXT

# (URL valid as of January 2008)

# The "end of central directory" structure, magic number, size, and indices

# (section V.I in the format document)

structEndArchive = b"<4s4H2LH"

stringEndArchive = b"PK\005\006"

sizeEndCentDir = struct.calcsize(structEndArchive)

_ECD_SIGNATURE = 0

_ECD_DISK_NUMBER = 1

_ECD_DISK_START = 2

_ECD_ENTRIES_THIS_DISK = 3

_ECD_ENTRIES_TOTAL = 4

_ECD_SIZE = 5

_ECD_OFFSET = 6

_ECD_COMMENT_SIZE = 7

# These last two indices are not part of the structure as defined in the

# spec, but they are used internally by this module as a convenience

_ECD_COMMENT = 8

_ECD_LOCATION = 9

# The "central directory" structure, magic number, size, and indices

# of entries in the structure (section V.F in the format document)

structCentralDir = "<4s4B4HL2L5H2L"

stringCentralDir = b"PK\001\002"

sizeCentralDir = struct.calcsize(structCentralDir)

# indexes of entries in the central directory structure

_CD_SIGNATURE = 0

_CD_CREATE_VERSION = 1

_CD_CREATE_SYSTEM = 2

_CD_EXTRACT_VERSION = 3

_CD_EXTRACT_SYSTEM = 4

_CD_FLAG_BITS = 5

_CD_COMPRESS_TYPE = 6

_CD_TIME = 7

_CD_DATE = 8

_CD_CRC = 9

_CD_COMPRESSED_SIZE = 10

_CD_UNCOMPRESSED_SIZE = 11

_CD_FILENAME_LENGTH = 12

_CD_EXTRA_FIELD_LENGTH = 13

_CD_COMMENT_LENGTH = 14

_CD_DISK_NUMBER_START = 15

_CD_INTERNAL_FILE_ATTRIBUTES = 16

_CD_EXTERNAL_FILE_ATTRIBUTES = 17

_CD_LOCAL_HEADER_OFFSET = 18

# General purpose bit flags

# Zip Appnote: 4.4.4 general purpose bit flag: (2 bytes)

_MASK_ENCRYPTED = 1 << 0

# Bits 1 and 2 have different meanings depending on the compression used.

_MASK_COMPRESS_OPTION_1 = 1 << 1

# _MASK_COMPRESS_OPTION_2 = 1 << 2

# _MASK_USE_DATA_DESCRIPTOR: If set, crc-32, compressed size and uncompressed

# size are zero in the local header and the real values are written in the data

# descriptor immediately following the compressed data.

_MASK_USE_DATA_DESCRIPTOR = 1 << 3

# Bit 4: Reserved for use with compression method 8, for enhanced deflating.

# _MASK_RESERVED_BIT_4 = 1 << 4

_MASK_COMPRESSED_PATCH = 1 << 5

_MASK_STRONG_ENCRYPTION = 1 << 6

# _MASK_UNUSED_BIT_7 = 1 << 7

# _MASK_UNUSED_BIT_8 = 1 << 8

# _MASK_UNUSED_BIT_9 = 1 << 9

# _MASK_UNUSED_BIT_10 = 1 << 10

_MASK_UTF_FILENAME = 1 << 11

# Bit 12: Reserved by PKWARE for enhanced compression.

# _MASK_RESERVED_BIT_12 = 1 << 12

# _MASK_ENCRYPTED_CENTRAL_DIR = 1 << 13

# Bit 14, 15: Reserved by PKWARE

# _MASK_RESERVED_BIT_14 = 1 << 14

# _MASK_RESERVED_BIT_15 = 1 << 15

# The "local file header" structure, magic number, size, and indices

# (section V.A in the format document)

structFileHeader = "<4s2B4HL2L2H"

stringFileHeader = b"PK\003\004"

sizeFileHeader = struct.calcsize(structFileHeader)

_FH_SIGNATURE = 0

_FH_EXTRACT_VERSION = 1

_FH_EXTRACT_SYSTEM = 2

_FH_GENERAL_PURPOSE_FLAG_BITS = 3

_FH_COMPRESSION_METHOD = 4

_FH_LAST_MOD_TIME = 5

_FH_LAST_MOD_DATE = 6

_FH_CRC = 7

_FH_COMPRESSED_SIZE = 8

_FH_UNCOMPRESSED_SIZE = 9

_FH_FILENAME_LENGTH = 10

_FH_EXTRA_FIELD_LENGTH = 11

# The "Zip64 end of central directory locator" structure, magic number, and size

structEndArchive64Locator = "<4sLQL"

stringEndArchive64Locator = b"PK\x06\x07"

sizeEndCentDir64Locator = struct.calcsize(structEndArchive64Locator)

# The "Zip64 end of central directory" record, magic number, size, and indices

# (section V.G in the format document)

structEndArchive64 = "<4sQ2H2L4Q"

stringEndArchive64 = b"PK\x06\x06"

sizeEndCentDir64 = struct.calcsize(structEndArchive64)

_CD64_SIGNATURE = 0

_CD64_DIRECTORY_RECSIZE = 1

_CD64_CREATE_VERSION = 2

_CD64_EXTRACT_VERSION = 3

_CD64_DISK_NUMBER = 4

_CD64_DISK_NUMBER_START = 5

_CD64_NUMBER_ENTRIES_THIS_DISK = 6

_CD64_NUMBER_ENTRIES_TOTAL = 7

_CD64_DIRECTORY_SIZE = 8

_CD64_OFFSET_START_CENTDIR = 9

_DD_SIGNATURE = 0x08074b50

class _Extra(bytes):

FIELD_STRUCT = struct.Struct('<HH')

def __new__(cls, val, id=None):

return super().__new__(cls, val)

def __init__(self, val, id=None):

self.id = id

@classmethod

def read_one(cls, raw):

try:

xid, xlen = cls.FIELD_STRUCT.unpack(raw[:4])

except struct.error:

xid = None

xlen = 0

return cls(raw[:4+xlen], xid), raw[4+xlen:]

@classmethod

def split(cls, data):

# use memoryview for zero-copy slices

rest = memoryview(data)

while rest:

extra, rest = _Extra.read_one(rest)

yield extra

@classmethod

def strip(cls, data, xids):

"""Remove Extra fields with specified IDs."""

return b''.join(

for ex in cls.split(data)

if ex.id not in xids

)

def _check_zipfile(fp):

try:

endrec = _EndRecData(fp)

if endrec:

if endrec[_ECD_ENTRIES_TOTAL] == 0 and endrec[_ECD_SIZE] == 0 and endrec[_ECD_OFFSET] == 0:

return True # Empty zipfiles are still zipfiles

elif endrec[_ECD_DISK_NUMBER] == endrec[_ECD_DISK_START]:

# Central directory is on the same disk

fp.seek(sum(_handle_prepended_data(endrec)))

if endrec[_ECD_SIZE] >= sizeCentralDir:

data = fp.read(sizeCentralDir) # CD is where we expect it to be

if len(data) == sizeCentralDir:

centdir = struct.unpack(structCentralDir, data) # CD is the right size

if centdir[_CD_SIGNATURE] == stringCentralDir:

return True # First central directory entry has correct magic number

except OSError:

pass

return False

def is_zipfile(filename):

"""Quickly see if a file is a ZIP file by checking the magic number.

The filename argument may be a file or file-like object too.

"""

result = False

try:

if hasattr(filename, "read"):

pos = filename.tell()

result = _check_zipfile(fp=filename)

filename.seek(pos)

else:

with open(filename, "rb") as fp:

result = _check_zipfile(fp)

except (OSError, BadZipFile):

pass

return result

def _handle_prepended_data(endrec, debug=0):

size_cd = endrec[_ECD_SIZE] # bytes in central directory

offset_cd = endrec[_ECD_OFFSET] # offset of central directory

# "concat" is zero, unless zip was concatenated to another file

concat = endrec[_ECD_LOCATION] - size_cd - offset_cd

if debug > 2:

inferred = concat + offset_cd

print("given, inferred, offset", offset_cd, inferred, concat)

return offset_cd, concat

def _EndRecData64(fpin, offset, endrec):

"""

Read the ZIP64 end-of-archive records and use that to update endrec

"""

offset -= sizeEndCentDir64Locator

if offset < 0:

# The file is not large enough to contain a ZIP64

# end-of-archive record, so just return the end record we were given.

return endrec

fpin.seek(offset)

data = fpin.read(sizeEndCentDir64Locator)

if len(data) != sizeEndCentDir64Locator:

raise OSError("Unknown I/O error")

sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data)

if sig != stringEndArchive64Locator:

return endrec

if diskno != 0 or disks > 1:

raise BadZipFile("zipfiles that span multiple disks are not supported")

offset -= sizeEndCentDir64

if reloff > offset:

raise BadZipFile("Corrupt zip64 end of central directory locator")

# First, check the assumption that there is no prepended data.

fpin.seek(reloff)

extrasz = offset - reloff

data = fpin.read(sizeEndCentDir64)

if len(data) != sizeEndCentDir64:

raise OSError("Unknown I/O error")

if not data.startswith(stringEndArchive64) and reloff != offset:

# Since we already have seen the Zip64 EOCD Locator, it's

# possible we got here because there is prepended data.

# Assume no 'zip64 extensible data'

fpin.seek(offset)

extrasz = 0

data = fpin.read(sizeEndCentDir64)

if len(data) != sizeEndCentDir64:

raise OSError("Unknown I/O error")

if not data.startswith(stringEndArchive64):

raise BadZipFile("Zip64 end of central directory record not found")

sig, sz, create_version, read_version, disk_num, disk_dir, \

dircount, dircount2, dirsize, diroffset = \

struct.unpack(structEndArchive64, data)

if (diroffset + dirsize != reloff or

sz + 12 != sizeEndCentDir64 + extrasz):

raise BadZipFile("Corrupt zip64 end of central directory record")

# Update the original endrec using data from the ZIP64 record

endrec[_ECD_SIGNATURE] = sig

endrec[_ECD_DISK_NUMBER] = disk_num

endrec[_ECD_DISK_START] = disk_dir

endrec[_ECD_ENTRIES_THIS_DISK] = dircount

endrec[_ECD_ENTRIES_TOTAL] = dircount2

endrec[_ECD_SIZE] = dirsize

endrec[_ECD_OFFSET] = diroffset

endrec[_ECD_LOCATION] = offset - extrasz

return endrec

def _EndRecData(fpin):

"""Return data from the "End of Central Directory" record, or None.

The data is a list of the nine items in the ZIP "End of central dir"

record followed by a tenth item, the file seek offset of this record."""

# Determine file size

fpin.seek(0, 2)

filesize = fpin.tell()

# Check to see if this is ZIP file with no archive comment (the

# "end of central directory" structure should be the last item in the

# file if this is the case).

try:

fpin.seek(-sizeEndCentDir, 2)

except OSError:

return None

data = fpin.read(sizeEndCentDir)

if (len(data) == sizeEndCentDir and

data[0:4] == stringEndArchive and

data[-2:] == b"\000\000"):

# the signature is correct and there's no comment, unpack structure

endrec = struct.unpack(structEndArchive, data)

endrec=list(endrec)

# Append a blank comment and record start offset

endrec.append(b"")

endrec.append(filesize - sizeEndCentDir)

# Try to read the "Zip64 end of central directory" structure

return _EndRecData64(fpin, filesize - sizeEndCentDir, endrec)

# Either this is not a ZIP file, or it is a ZIP file with an archive

# comment. Search the end of the file for the "end of central directory"

# record signature. The comment is the last item in the ZIP file and may be

# up to 64K long. It is assumed that the "end of central directory" magic

# number does not appear in the comment.

maxCommentStart = max(filesize - ZIP_MAX_COMMENT - sizeEndCentDir, 0)

fpin.seek(maxCommentStart, 0)

data = fpin.read(ZIP_MAX_COMMENT + sizeEndCentDir)

start = data.rfind(stringEndArchive)

if start >= 0:

# found the magic number; attempt to unpack and interpret

recData = data[start:start+sizeEndCentDir]

if len(recData) != sizeEndCentDir:

# Zip file is corrupted.

return None

endrec = list(struct.unpack(structEndArchive, recData))

commentSize = endrec[_ECD_COMMENT_SIZE] #as claimed by the zip file

comment = data[start+sizeEndCentDir:start+sizeEndCentDir+commentSize]

endrec.append(comment)

endrec.append(maxCommentStart + start)

# Try to read the "Zip64 end of central directory" structure

return _EndRecData64(fpin, maxCommentStart + start, endrec)

# Unable to find a valid end of central directory structure

return None

def _sanitize_filename(filename):

"""Terminate the file name at the first null byte and

ensure paths always use forward slashes as the directory separator."""

# Terminate the file name at the first null byte. Null bytes in file

# names are used as tricks by viruses in archives.

null_byte = filename.find(chr(0))

if null_byte >= 0:

filename = filename[0:null_byte]

# This is used to ensure paths in generated ZIP files always use

# forward slashes as the directory separator, as required by the

# ZIP format specification.

if os.sep != "/" and os.sep in filename:

filename = filename.replace(os.sep, "/")

if os.altsep and os.altsep != "/" and os.altsep in filename:

filename = filename.replace(os.altsep, "/")

return filename

def _read_local_file_header(fp):

fheader = fp.read(sizeFileHeader)

if len(fheader) != sizeFileHeader:

raise BadZipFile("Truncated file header")

fheader = struct.unpack(structFileHeader, fheader)

if fheader[_FH_SIGNATURE] != stringFileHeader:

raise BadZipFile("Bad magic number for file header")

return fheader

class ZipInfo:

"""Class with attributes describing each file in the ZIP archive."""

__slots__ = (

'orig_filename',

'filename',

'date_time',

'compress_type',

'compress_level',

'comment',

'extra',

'create_system',

'create_version',

'extract_version',

'reserved',

'flag_bits',

'volume',

'internal_attr',

'external_attr',

'header_offset',

'CRC',

'compress_size',

'file_size',

'_raw_time',

'_end_offset',

)

def __init__(self, filename="NoName", date_time=(1980,1,1,0,0,0)):

self.orig_filename = filename # Original file name in archive

# Terminate the file name at the first null byte and

# ensure paths always use forward slashes as the directory separator.

filename = _sanitize_filename(filename)

self.filename = filename # Normalized file name

self.date_time = date_time # year, month, day, hour, min, sec

if date_time[0] < 1980:

raise ValueError('ZIP does not support timestamps before 1980')

# Standard values:

self.compress_type = ZIP_STORED # Type of compression for the file

self.compress_level = None # Level for the compressor

self.comment = b"" # Comment for each file

self.extra = b"" # ZIP extra data

if sys.platform == 'win32':

self.create_system = 0 # System which created ZIP archive

else:

# Assume everything else is unix-y

self.create_system = 3 # System which created ZIP archive

self.create_version = DEFAULT_VERSION # Version which created ZIP archive

self.extract_version = DEFAULT_VERSION # Version needed to extract archive

self.reserved = 0 # Must be zero

self.flag_bits = 0 # ZIP flag bits

self.volume = 0 # Volume number of file header

self.internal_attr = 0 # Internal attributes

self.external_attr = 0 # External file attributes

self.compress_size = 0 # Size of the compressed file

self.file_size = 0 # Size of the uncompressed file

self._end_offset = None # Start of the next local header or central directory

# Other attributes are set by class ZipFile:

# header_offset Byte offset to the file header

# CRC CRC-32 of the uncompressed file

# Maintain backward compatibility with the old protected attribute name.

@property

def _compresslevel(self):

return self.compress_level

@_compresslevel.setter

def _compresslevel(self, value):

self.compress_level = value

def __repr__(self):

result = ['<%s filename=%r' % (self.__class__.__name__, self.filename)]

if self.compress_type != ZIP_STORED:

result.append(' compress_type=%s' %

compressor_names.get(self.compress_type,

self.compress_type))

hi = self.external_attr >> 16

lo = self.external_attr & 0xFFFF

if hi:

result.append(' filemode=%r' % stat.filemode(hi))

if lo:

result.append(' external_attr=%#x' % lo)

isdir = self.is_dir()

if not isdir or self.file_size:

result.append(' file_size=%r' % self.file_size)

if ((not isdir or self.compress_size) and

(self.compress_type != ZIP_STORED or

self.file_size != self.compress_size)):

result.append(' compress_size=%r' % self.compress_size)

result.append('>')

return ''.join(result)

def FileHeader(self, zip64=None):

"""Return the per-file header as a bytes object.

When the optional zip64 arg is None rather than a bool, we will

decide based upon the file_size and compress_size, if known,

False otherwise.

"""

dt = self.date_time

dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2]

dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2)

if self.flag_bits & _MASK_USE_DATA_DESCRIPTOR:

# Set these to zero because we write them after the file data

CRC = compress_size = file_size = 0

else:

CRC = self.CRC

compress_size = self.compress_size

file_size = self.file_size

extra = self.extra

min_version = 0

if zip64 is None:

# We always explicitly pass zip64 within this module.... This

# remains for anyone using ZipInfo.FileHeader as a public API.

zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT

if zip64:

fmt = '<HHQQ'

extra = extra + struct.pack(fmt,

1, struct.calcsize(fmt)-4, file_size, compress_size)

file_size = 0xffffffff

compress_size = 0xffffffff

min_version = ZIP64_VERSION

if self.compress_type == ZIP_BZIP2:

min_version = max(BZIP2_VERSION, min_version)

elif self.compress_type == ZIP_LZMA:

min_version = max(LZMA_VERSION, min_version)

elif self.compress_type == ZIP_ZSTANDARD:

min_version = max(ZSTANDARD_VERSION, min_version)

self.extract_version = max(min_version, self.extract_version)

self.create_version = max(min_version, self.create_version)

filename, flag_bits = self._encodeFilenameFlags()

header = struct.pack(structFileHeader, stringFileHeader,

self.extract_version, self.reserved, flag_bits,

self.compress_type, dostime, dosdate, CRC,

compress_size, file_size,

len(filename), len(extra))

return header + filename + extra

def _encodeFilenameFlags(self):

if self.flag_bits & _MASK_UTF_FILENAME:

encoding = 'ascii'

else:

encoding = 'cp437'

try:

return self.filename.encode(encoding), self.flag_bits & ~_MASK_UTF_FILENAME

except UnicodeEncodeError:

return self.filename.encode('utf-8'), self.flag_bits | _MASK_UTF_FILENAME

def _decodeExtra(self, filename_crc):

# Try to decode the extra field.

extra = self.extra

unpack = struct.unpack

while len(extra) >= 4:

tp, ln = unpack('<HH', extra[:4])

if ln+4 > len(extra):

raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln))

if tp == 0x0001:

data = extra[4:ln+4]

# ZIP64 extension (large files and/or large archives)

try:

if self.file_size in (0xFFFF_FFFF_FFFF_FFFF, 0xFFFF_FFFF):

field = "File size"

self.file_size, = unpack('<Q', data[:8])

data = data[8:]

if self.compress_size == 0xFFFF_FFFF:

field = "Compress size"

self.compress_size, = unpack('<Q', data[:8])

data = data[8:]

if self.header_offset == 0xFFFF_FFFF:

field = "Header offset"

self.header_offset, = unpack('<Q', data[:8])

except struct.error:

raise BadZipFile(f"Corrupt zip64 extra field. "

f"{field} not found.") from None

elif tp == 0x7075:

data = extra[4:ln+4]

# Unicode Path Extra Field

try:

up_version, up_name_crc = unpack('<BL', data[:5])

if up_version == 1 and up_name_crc == filename_crc:

up_unicode_name = data[5:].decode('utf-8')

if up_unicode_name:

self.filename = _sanitize_filename(up_unicode_name)

else:

warnings.warn("Empty unicode path extra field (0x7075)", stacklevel=2)

except struct.error as e:

raise BadZipFile("Corrupt unicode path extra field (0x7075)") from e

except UnicodeDecodeError as e:

raise BadZipFile('Corrupt unicode path extra field (0x7075): invalid utf-8 bytes') from e

extra = extra[ln+4:]

@classmethod

def from_file(cls, filename, arcname=None, *, strict_timestamps=True):

"""Construct an appropriate ZipInfo for a file on the filesystem.

filename should be the path to a file or directory on the

filesystem.

arcname is the name which it will have within the archive (by

default, this will be the same as filename, but without a drive

letter and with leading path separators removed).

"""

if isinstance(filename, os.PathLike):

filename = os.fspath(filename)

st = os.stat(filename)

isdir = stat.S_ISDIR(st.st_mode)

mtime = time.localtime(st.st_mtime)

date_time = mtime[0:6]

if not strict_timestamps and date_time[0] < 1980:

date_time = (1980, 1, 1, 0, 0, 0)

elif not strict_timestamps and date_time[0] > 2107:

date_time = (2107, 12, 31, 23, 59, 59)

# Create ZipInfo instance to store file information

if arcname is None:

arcname = filename

arcname = os.path.normpath(os.path.splitdrive(arcname)[1])

while arcname[0] in (os.sep, os.altsep):

arcname = arcname[1:]

if isdir:

arcname += '/'

zinfo = cls(arcname, date_time)

zinfo.external_attr = (st.st_mode & 0xFFFF) << 16 # Unix attributes

if isdir:

zinfo.file_size = 0

zinfo.external_attr |= 0x10 # MS-DOS directory flag

else:

zinfo.file_size = st.st_size

return zinfo

def _for_archive(self, archive):

"""Resolve suitable defaults from the archive.

Resolve the date_time, compression attributes, and external attributes

to suitable defaults as used by :method:`ZipFile.writestr`.

Return self.

"""

# gh-91279: Set the SOURCE_DATE_EPOCH to a specific timestamp

source_date_epoch = os.environ.get('SOURCE_DATE_EPOCH')

if source_date_epoch:

self.date_time = time.gmtime(int(source_date_epoch))[:6]

else:

self.date_time = time.localtime(time.time())[:6]

self.compress_type = archive.compression

self.compress_level = archive.compresslevel

if self.filename.endswith('/'): # pragma: no cover

self.external_attr = 0o40775 << 16 # drwxrwxr-x

self.external_attr |= 0x10 # MS-DOS directory flag

else:

self.external_attr = 0o100600 << 16 # -rw-------

return self

def is_dir(self):

"""Return True if this archive member is a directory."""

if self.filename.endswith('/'):

return True

# The ZIP format specification requires to use forward slashes

# as the directory separator, but in practice some ZIP files

# created on Windows can use backward slashes. For compatibility

# with the extraction code which already handles this:

if os.path.altsep:

return self.filename.endswith((os.path.sep, os.path.altsep))

return False

# ZIP encryption uses the CRC32 one-byte primitive for scrambling some

# internal keys. We noticed that a direct implementation is faster than

# relying on binascii.crc32().

_crctable = None

def _gen_crc(crc):

for j in range(8):

if crc & 1:

crc = (crc >> 1) ^ 0xEDB88320

else:

crc >>= 1

return crc

# ZIP supports a password-based form of encryption. Even though known

# plaintext attacks have been found against it, it is still useful

# to be able to get data out of such a file.

# Usage:

# zd = _ZipDecrypter(mypwd)

# plain_bytes = zd(cypher_bytes)

def _ZipDecrypter(pwd):

key0 = 305419896

key1 = 591751049

key2 = 878082192

global _crctable

if _crctable is None:

_crctable = list(map(_gen_crc, range(256)))

crctable = _crctable

def crc32(ch, crc):

"""Compute the CRC32 primitive on one byte."""

return (crc >> 8) ^ crctable[(crc ^ ch) & 0xFF]

def update_keys(c):

nonlocal key0, key1, key2

key0 = crc32(c, key0)

key1 = (key1 + (key0 & 0xFF)) & 0xFFFFFFFF

key1 = (key1 * 134775813 + 1) & 0xFFFFFFFF

key2 = crc32(key1 >> 24, key2)

for p in pwd:

update_keys(p)

def decrypter(data):

"""Decrypt a bytes object."""

result = bytearray()

append = result.append

for c in data:

k = key2 | 2

c ^= ((k * (k^1)) >> 8) & 0xFF

update_keys(c)

append(c)

return bytes(result)

return decrypter

class LZMACompressor:

def __init__(self):

self._comp = None

def _init(self):

props = lzma._encode_filter_properties({'id': lzma.FILTER_LZMA1})

self._comp = lzma.LZMACompressor(lzma.FORMAT_RAW, filters=[

lzma._decode_filter_properties(lzma.FILTER_LZMA1, props)

])

return struct.pack('<BBH', 9, 4, len(props)) + props

def compress(self, data):

if self._comp is None:

return self._init() + self._comp.compress(data)

return self._comp.compress(data)

def flush(self):

if self._comp is None:

return self._init() + self._comp.flush()

return self._comp.flush()

class LZMADecompressor:

def __init__(self):

self._decomp = None

self._unconsumed = b''

self.eof = False

@property

def unused_data(self):

try:

return self._decomp.unused_data

except AttributeError:

return b''

@property

def needs_input(self):

# While the LZMA properties header is still being buffered, more input

# is required; afterwards defer to the wrapped decompressor so a bounded

# decompress() call can be drained across reads.

if self._decomp is None:

return True

return self._decomp.needs_input

def decompress(self, data, max_length=-1):

if self._decomp is None:

self._unconsumed += data

if len(self._unconsumed) <= 4:

return b''

psize, = struct.unpack('<H', self._unconsumed[2:4])

if len(self._unconsumed) <= 4 + psize:

return b''

self._decomp = lzma.LZMADecompressor(lzma.FORMAT_RAW, filters=[

lzma._decode_filter_properties(lzma.FILTER_LZMA1,

self._unconsumed[4:4 + psize])

])

data = self._unconsumed[4 + psize:]

del self._unconsumed

result = self._decomp.decompress(data, max_length)

self.eof = self._decomp.eof

return result

compressor_names = {

0: 'store',

1: 'shrink',

2: 'reduce',

3: 'reduce',

4: 'reduce',

5: 'reduce',

6: 'implode',

7: 'tokenize',

8: 'deflate',

9: 'deflate64',

10: 'implode',

12: 'bzip2',

14: 'lzma',

18: 'terse',

19: 'lz77',

93: 'zstd',

97: 'wavpack',

98: 'ppmd',

}

def _check_compression(compression):

if compression == ZIP_STORED:

pass

elif compression == ZIP_DEFLATED:

if not zlib:

raise RuntimeError(

"Compression requires the (missing) zlib module")

elif compression == ZIP_BZIP2:

if not bz2:

raise RuntimeError(

"Compression requires the (missing) bz2 module")

elif compression == ZIP_LZMA:

if not lzma:

raise RuntimeError(

"Compression requires the (missing) lzma module")

elif compression == ZIP_ZSTANDARD:

if not zstd:

raise RuntimeError(

"Compression requires the (missing) compression.zstd module")

else:

raise NotImplementedError("That compression method is not supported")

def _get_compressor(compress_type, compresslevel=None):

if compress_type == ZIP_DEFLATED:

if compresslevel is not None:

return zlib.compressobj(compresslevel, zlib.DEFLATED, -15)

return zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -15)

elif compress_type == ZIP_BZIP2:

if compresslevel is not None:

return bz2.BZ2Compressor(compresslevel)

return bz2.BZ2Compressor()

# compresslevel is ignored for ZIP_LZMA

elif compress_type == ZIP_LZMA:

return LZMACompressor()

elif compress_type == ZIP_ZSTANDARD:

return zstd.ZstdCompressor(level=compresslevel)

else:

return None

def _get_decompressor(compress_type):

_check_compression(compress_type)

if compress_type == ZIP_STORED:

return None

elif compress_type == ZIP_DEFLATED:

return zlib.decompressobj(-15)

elif compress_type == ZIP_BZIP2:

return bz2.BZ2Decompressor()

elif compress_type == ZIP_LZMA:

return LZMADecompressor()

elif compress_type == ZIP_ZSTANDARD:

return zstd.ZstdDecompressor()

else:

descr = compressor_names.get(compress_type)

if descr:

raise NotImplementedError("compression type %d (%s)" % (compress_type, descr))

else:

raise NotImplementedError("compression type %d" % (compress_type,))

class _SharedFile:

def __init__(self, file, pos, close, lock, writing):

self._file = file

self._pos = pos

self._close = close

self._lock = lock

self._writing = writing

self.seekable = file.seekable

def tell(self):

return self._pos

def seek(self, offset, whence=0):

with self._lock:

if self._writing():

raise ValueError("Can't reposition in the ZIP file while "

"there is an open writing handle on it. "

"Close the writing handle before trying to read.")

if whence == os.SEEK_CUR:

self._file.seek(self._pos + offset)

else:

self._file.seek(offset, whence)

self._pos = self._file.tell()

return self._pos

def read(self, n=-1):

with self._lock:

if self._writing():

raise ValueError("Can't read from the ZIP file while there "

"is an open writing handle on it. "

"Close the writing handle before trying to read.")

self._file.seek(self._pos)

data = self._file.read(n)

self._pos = self._file.tell()

return data

def close(self):

if self._file is not None:

fileobj = self._file

self._file = None

self._close(fileobj)

# Provide the tell method for unseekable stream

class _Tellable:

def __init__(self, fp):

self.fp = fp

self.offset = 0

def write(self, data):

n = self.fp.write(data)

self.offset += n

return n

def tell(self):

return self.offset

def flush(self):

self.fp.flush()

def close(self):

self.fp.close()

class ZipExtFile(io.BufferedIOBase):

"""File-like object for reading an archive member.

Is returned by ZipFile.open().

"""

# Max size supported by decompressor.

MAX_N = (1 << 31) - 1

# Read from compressed files in 4k blocks.

MIN_READ_SIZE = 4096

# Chunk size to read during seek

MAX_SEEK_READ = 1 << 24

def __init__(self, fileobj, mode, zipinfo, pwd=None,

close_fileobj=False):

self._fileobj = fileobj

self._pwd = pwd

self._close_fileobj = close_fileobj

self._compress_type = zipinfo.compress_type

self._compress_left = zipinfo.compress_size

View remainder of file in raw view

cpython/Lib/zipfile/init.py at main · python/cpython · GitHub

Repository navigation

init.py

init.py

cpython/Lib/zipfile/__init__.py at main · python/cpython · GitHub

Repository navigation

Expand file tree

Breadcrumbs

__init__.py

Latest commit

History

Breadcrumbs

__init__.py

File metadata and controls

cpython/Lib/zipfile/init.py at main · python/cpython · GitHub

init.py

init.py