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 contextlib
2import os
3import re
4import tarfile
5from pathlib import Path
7from structlog import get_logger
9from ...file_utils import OffsetFile, SeekError, decode_int, round_up, snull
10from ...models import (
11 Extractor,
12 ExtractResult,
13 File,
14 HandlerDoc,
15 HandlerType,
16 HexString,
17 Reference,
18 Regex,
19 StructHandler,
20 ValidChunk,
21)
22from ._safe_tarfile import SafeTarFile, open_safe_tarfile
24logger = get_logger()
27BLOCK_SIZE = 512
28END_OF_ARCHIVE_MARKER_SIZE = 2 * BLOCK_SIZE
30MAGIC_OFFSET = 257
32ZERO_BLOCK = bytes([0]) * BLOCK_SIZE
35def _get_tar_end_offset(file: File, offset=0):
36 file_with_offset = OffsetFile(file, offset)
38 # First find the end of the last entry in the file
39 last_offset = _get_end_of_last_tar_entry(file_with_offset)
40 if last_offset == -1:
41 return -1
43 # Then find where the final zero blocks end
44 return offset + _find_end_of_padding(file_with_offset, find_from=last_offset)
47def _get_end_of_last_tar_entry(file) -> int:
48 try:
49 tf = open_safe_tarfile(mode="r", fileobj=file)
50 except tarfile.TarError:
51 return -1
53 last_member = None
55 try:
56 for member in tf:
57 last_member = member
58 except (tarfile.TarError, SeekError):
59 # recover what's already been parsed
60 pass
62 if last_member is None:
63 return -1
65 end_of_last_tar_entry = tf.offset
66 try:
67 file.seek(end_of_last_tar_entry)
68 except SeekError:
69 # last tar entry is truncated
70 end_of_last_tar_entry = last_member.offset
71 file.seek(end_of_last_tar_entry)
73 return end_of_last_tar_entry
76def _find_end_of_padding(file, *, find_from: int) -> int:
77 find_from = round_up(find_from, BLOCK_SIZE)
78 find_to = round_up(find_from + END_OF_ARCHIVE_MARKER_SIZE, tarfile.RECORDSIZE)
80 max_padding_blocks = (find_to - find_from) // BLOCK_SIZE
82 try:
83 file.seek(find_from)
84 except SeekError:
85 # match to end of truncated file
86 return file.seek(0, os.SEEK_END)
88 for padding_blocks in range(max_padding_blocks): # noqa: B007
89 if file.read(BLOCK_SIZE) != ZERO_BLOCK:
90 break
91 else:
92 padding_blocks = max_padding_blocks
94 return find_from + padding_blocks * BLOCK_SIZE
97class TarExtractor(Extractor):
98 def extract(self, inpath: Path, outdir: Path):
99 with contextlib.closing(SafeTarFile(inpath)) as tarfile:
100 tarfile.extractall(outdir) # noqa: S202 tarfile-unsafe-members
101 return ExtractResult(reports=tarfile.reports)
104class _TarHandler(StructHandler):
105 NAME = "tar"
107 PATTERNS = []
109 C_DEFINITIONS = r"""
110 typedef struct posix_header
111 { /* byte offset */
112 char name[100]; /* 0 */
113 char mode[8]; /* 100 */
114 char uid[8]; /* 108 */
115 char gid[8]; /* 116 */
116 char file_size[12]; /* 124 */
117 char mtime[12]; /* 136 */
118 char chksum[8]; /* 148 */
119 char typeflag; /* 156 */
120 char linkname[100]; /* 157 */
121 char magic[6]; /* 257 */
122 char version[2]; /* 263 */
123 char uname[32]; /* 265 */
124 char gname[32]; /* 297 */
125 char devmajor[8]; /* 329 */
126 char devminor[8]; /* 337 */
127 char prefix[155]; /* 345 */
128 /* 500 */
129 } posix_header_t;
130 """
131 HEADER_STRUCT = "posix_header_t"
133 EXTRACTOR = TarExtractor()
135 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
136 file.seek(start_offset)
137 header = self.parse_header(file)
138 header_size = snull(header.file_size)
139 decode_int(header_size, 8)
141 def signed_sum(octets) -> int:
142 return sum(b if b < 128 else 256 - b for b in octets)
144 chksum_match = re.search(rb"^(?P<digits>[0-7]+)[\x00\x20]+$", header.chksum)
145 if not chksum_match:
146 logger.debug(
147 "Invalid checksum format",
148 chksum=header.chksum,
149 handler=self.NAME,
150 _verbosity=3,
151 )
152 return None
153 checksum = decode_int(chksum_match.group("digits"), 8)
154 header_bytes_for_checksum = (
155 file[start_offset : start_offset + 148]
156 + b" " * 8 # chksum field is replaced with "blanks"
157 + file[start_offset + 156 : start_offset + 257]
158 )
159 extended_header_bytes = file[start_offset + 257 : start_offset + 500]
160 calculated_checksum_unsigned = sum(header_bytes_for_checksum)
161 calculated_checksum_signed = signed_sum(header_bytes_for_checksum)
162 checksums = (
163 calculated_checksum_unsigned,
164 calculated_checksum_unsigned + sum(extended_header_bytes),
165 # signed is of historical interest, calculating for the extended header is not needed
166 calculated_checksum_signed,
167 )
168 if checksum not in checksums:
169 logger.error(
170 "Tar header checksum mismatch", expected=str(checksum), actual=checksums
171 )
172 return None
174 end_offset = _get_tar_end_offset(file, start_offset)
175 if end_offset == -1:
176 return None
177 return ValidChunk(start_offset=start_offset, end_offset=end_offset)
180class TarUstarHandler(_TarHandler):
181 PATTERNS = [
182 HexString("75 73 74 61 72 20 20 00"),
183 HexString("75 73 74 61 72 00 30 30"),
184 ]
186 # Since the magic is at 257, we have to subtract that from the match offset
187 # to get to the start of the file.
188 PATTERN_MATCH_OFFSET = -MAGIC_OFFSET
190 DOC = HandlerDoc(
191 name="TAR (USTAR)",
192 description="USTAR (Uniform Standard Tape Archive) tar files are extensions of the original tar format with additional metadata fields.",
193 handler_type=HandlerType.ARCHIVE,
194 vendor=None,
195 references=[
196 Reference(
197 title="USTAR Format Documentation",
198 url="https://en.wikipedia.org/wiki/Tar_(computing)#USTAR_format",
199 ),
200 Reference(
201 title="POSIX Tar Format Specification",
202 url="https://pubs.opengroup.org/onlinepubs/9699919799/utilities/pax.html",
203 ),
204 ],
205 limitations=[],
206 )
209def _re_frame(regexp: str):
210 """Wrap regexp to ensure its integrity from concatenation.
212 E.g.: when the regex
213 a|b
214 is naively appended by regex c, the result
215 a|bc
216 will not match "ac", while
217 (a|b)c
218 will match "ac" as intended.
219 """
220 return f"({regexp})"
223def _re_alternatives(regexps):
224 return _re_frame("|".join(_re_frame(regexp) for regexp in regexps))
227def _padded_field(re_content_char, size, leftpad_re=" ", rightpad_re=r"[ \0x00]"):
228 field_regexes = []
230 for padsize in range(size):
231 content_re = f"{re_content_char}{{{size - padsize}}}"
233 for leftpadsize in range(padsize + 1):
234 rightpadsize = padsize - leftpadsize
236 left_re = f"{leftpad_re}{{{leftpadsize}}}" if leftpadsize else ""
237 right_re = f"{rightpad_re}{{{rightpadsize}}}" if rightpadsize else ""
239 field_regexes.append(f"{left_re}{content_re}{right_re}")
241 return _re_alternatives(field_regexes)
244class TarUnixHandler(_TarHandler):
245 PATTERNS = [
246 Regex(
247 r""
248 # (pattern would be too big) char name[100]
249 + _padded_field(r"[0-7]", 8) # char mode[8]
250 + _padded_field(r"[0-7]", 8) # char uid[8]
251 + _padded_field(r"[0-7]", 8) # char gid[8]
252 + _padded_field(r"[0-7]", 12) # char file_size[12]
253 + _padded_field(r"[0-7]", 12) # char mtime[12]
254 + _padded_field(r"[0-7]", 8) # char chksum[8]
255 + r"[0-7\x00]" # char typeflag[1] - no extensions
256 ),
257 ]
258 PATTERN_MATCH_OFFSET = -100 # go back to beginning of skipped name
260 DOC = HandlerDoc(
261 name="TAR (Unix)",
262 description="Unix tar files are a widely used archive format for storing files and directories with metadata.",
263 handler_type=HandlerType.ARCHIVE,
264 vendor=None,
265 references=[
266 Reference(
267 title="Unix Tar Format Documentation",
268 url="https://en.wikipedia.org/wiki/Tar_(computing)",
269 ),
270 Reference(
271 title="GNU Tar Manual",
272 url="https://www.gnu.org/software/tar/manual/",
273 ),
274 ],
275 limitations=[],
276 )