Repository navigation
Expand file tree
/
Copy pathxtzip.py
More file actions
121 lines (111 loc) · 4.29 KB
/
Copy pathxtzip.py
File metadata and controls
121 lines (111 loc) · 4.29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from __future__ import annotations
import codecs
from refinery.lib.id import buffer_offset, is_likely_pe
from refinery.lib.types import buf
from refinery.lib.zip import (
InvalidChecksum,
InvalidPassword,
PasswordRequired,
UnsupportedCompressionMethod,
Zip,
ZipDirEntry,
ZipEndOfCentralDirectory,
ZipEndOfCentralDirectory64,
)
from refinery.units import RefineryPartialResult
from refinery.units.formats.archive import ArchiveUnit, MultipleArchives
from refinery.units.formats.pe import get_pe_size
class xtzip(ArchiveUnit, docs='{0}{p}{PathExtractorUnit}'):
"""
Extract files from a Zip archive.
"""
def unpack(self, data: buf):
def trypwd(password: str | None):
try:
zipf = Zip(view, password)
except (PasswordRequired, InvalidPassword):
return None
for file in zipf.records.values():
if file.is_dir():
continue
if file.is_password_ok(password):
break
return False
return zipf
view = memoryview(data)
password = self.args.pwd
if not password:
password = None
elif not isinstance(password, str):
password = codecs.decode(password, self.codec)
passwords = [password]
if not password:
passwords.extend(self.CommonPasswords)
for p in passwords:
if zipf := trypwd(p):
break
else:
zipf = Zip(view, password)
if some := zipf.sub_archive_count() and not self.args.lenient:
text = (
F'The input contains {some + 1} archives. Use the xtzip unit to extract '
R'them individually or set the --lenient/-L option to fuse the archives.')
raise MultipleArchives(text)
if zipf.password:
self.log_debug('Using password:', zipf.password)
if boundary := zipf.coverage.boundary():
w = len(hex(boundary[1]))
for start, end in zipf.coverage.gaps():
self.log_info(F'data cave detected at range {start:#0{w}x}:{end:#0{w}x}')
yield self._pack(F'.{start:#0{w}x}.cave', None, view[start:end])
for entry in sorted(zipf.directory, key=lambda d: d.name):
def xt(entry=entry):
record = zipf.read(entry)
try:
return record.unpack(zipf.password)
except InvalidChecksum as ck:
raise RefineryPartialResult('invalid checksum', ck.data) from ck
except (PasswordRequired, InvalidPassword):
if not record.data:
raise
msg = 'invalid password; use -L to extract raw encrypted data'
raise RefineryPartialResult(msg, record.data)
except UnsupportedCompressionMethod as error:
msg = F'{error} Use -L to extract the compressed data.'
raise RefineryPartialResult(msg, error.data) from error
if entry.is_dir():
continue
yield self._pack(entry.name, entry.date, xt)
@classmethod
def handles(cls, data):
if data[:4] in (
B'PK\x03\x04',
B'PK\x07\x08',
):
return True
for EOCD in (
ZipEndOfCentralDirectory64,
ZipEndOfCentralDirectory,
):
if buffer_offset(data, EOCD.Signature, back2front=True) > 0:
return True
if not is_likely_pe(data):
return False
memory = memoryview(data)
if 0 <= buffer_offset(memory[-0x400:], ZipDirEntry.Signature):
return True
from refinery.lib import lief
pe = lief.load_pe_fast(data)
offset = get_pe_size(pe)
if 0 <= buffer_offset(memory[offset:], B'PK\x03\x04') < 0x1000:
return True
if not pe.has_debug:
return False
for entry in pe.debug:
if not isinstance(entry, lief.PE.CodeViewPDB):
continue
path = entry.filename
if not isinstance(path, str):
path = codecs.decode(path, 'latin1')
if 'sfxzip32' in path and 'WinRAR' in path:
return True