Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/handlers/archive/_safe_tarfile.py: 63%

1import os 

2import tarfile 

3from pathlib import Path 

4from typing import Literal 

5 

6from structlog import get_logger 

7 

8from unblob.file_utils import is_safe_path 

9from unblob.report import ExtractionProblem 

10 

11logger = get_logger() 

12 

13RUNNING_AS_ROOT = os.getuid() == 0 

14MAX_PATH_LEN = 255 

15 

16 

17class UnblobTarInfo(tarfile.TarInfo): 

18 @classmethod 

19 def frombuf(cls, buf, encoding, errors): 

20 """Parse GNU headers without treating the prefix field as a pathname.""" 

21 return cls._frombuf(buf, encoding, errors) 

22 

23 @classmethod 

24 def _frombuf(cls, buf, encoding, errors, *, dircheck=True): # noqa: C901 

25 if len(buf) == 0: 

26 raise tarfile.EmptyHeaderError("empty header") # pyright: ignore[reportAttributeAccessIssue] 

27 if len(buf) != tarfile.BLOCKSIZE: 

28 raise tarfile.TruncatedHeaderError("truncated header") # pyright: ignore[reportAttributeAccessIssue] 

29 if buf.count(tarfile.NUL) == tarfile.BLOCKSIZE: 

30 raise tarfile.EOFHeaderError("end of file header") # pyright: ignore[reportAttributeAccessIssue] 

31 

32 chksum = tarfile.nti(buf[148:156]) # pyright: ignore[reportAttributeAccessIssue] 

33 if chksum not in tarfile.calc_chksums(buf): # pyright: ignore[reportAttributeAccessIssue] 

34 raise tarfile.InvalidHeaderError("bad checksum") # pyright: ignore[reportAttributeAccessIssue] 

35 

36 obj = cls() 

37 obj.name = tarfile.nts(buf[0:100], encoding, errors) # pyright: ignore[reportAttributeAccessIssue] 

38 obj.mode = tarfile.nti(buf[100:108]) # pyright: ignore[reportAttributeAccessIssue] 

39 obj.uid = tarfile.nti(buf[108:116]) # pyright: ignore[reportAttributeAccessIssue] 

40 obj.gid = tarfile.nti(buf[116:124]) # pyright: ignore[reportAttributeAccessIssue] 

41 obj.size = tarfile.nti(buf[124:136]) # pyright: ignore[reportAttributeAccessIssue] 

42 obj.mtime = tarfile.nti(buf[136:148]) # pyright: ignore[reportAttributeAccessIssue] 

43 obj.chksum = chksum 

44 obj.type = bytes(buf[156:157]) 

45 obj.linkname = tarfile.nts(buf[157:257], encoding, errors) # pyright: ignore[reportAttributeAccessIssue] 

46 obj.uname = tarfile.nts(buf[265:297], encoding, errors) # pyright: ignore[reportAttributeAccessIssue] 

47 obj.gname = tarfile.nts(buf[297:329], encoding, errors) # pyright: ignore[reportAttributeAccessIssue] 

48 obj.devmajor = tarfile.nti(buf[329:337]) # pyright: ignore[reportAttributeAccessIssue] 

49 obj.devminor = tarfile.nti(buf[337:345]) # pyright: ignore[reportAttributeAccessIssue] 

50 prefix = tarfile.nts(buf[345:500], encoding, errors) # pyright: ignore[reportAttributeAccessIssue] 

51 magic = buf[257:265] 

52 

53 if dircheck and obj.type == tarfile.AREGTYPE and obj.name.endswith("/"): 

54 obj.type = tarfile.DIRTYPE 

55 

56 if obj.type == tarfile.GNUTYPE_SPARSE: 

57 pos = 386 

58 structs = [] 

59 for _ in range(4): 

60 try: 

61 offset = tarfile.nti(buf[pos : pos + 12]) # pyright: ignore[reportAttributeAccessIssue] 

62 numbytes = tarfile.nti(buf[pos + 12 : pos + 24]) # pyright: ignore[reportAttributeAccessIssue] 

63 except ValueError: 

64 break 

65 structs.append((offset, numbytes)) 

66 pos += 24 

67 isextended = bool(buf[482]) 

68 origsize = tarfile.nti(buf[483:495]) # pyright: ignore[reportAttributeAccessIssue] 

69 obj._sparse_structs = (structs, isextended, origsize) 

70 

71 if obj.isdir(): 

72 obj.name = obj.name.rstrip("/") 

73 

74 if ( 

75 prefix 

76 and magic == tarfile.POSIX_MAGIC 

77 and obj.type not in tarfile.GNU_TYPES 

78 ): 

79 obj.name = prefix + "/" + obj.name 

80 return obj 

81 

82 

83def open_safe_tarfile( 

84 name=None, 

85 mode: Literal["r", "r:*", "r:", "r:gz", "r:bz2", "r:xz"] = "r", 

86 fileobj=None, 

87 **kwargs, 

88) -> tarfile.TarFile: 

89 return tarfile.open( # pyright: ignore[reportCallIssue] 

90 name=name, 

91 mode=mode, 

92 fileobj=fileobj, 

93 tarinfo=UnblobTarInfo, 

94 **kwargs, 

95 ) 

96 

97 

98class SafeTarFile: 

99 def __init__(self, inpath: Path): 

100 self.inpath = inpath 

101 self.reports = [] 

102 self.tarfile = open_safe_tarfile(inpath) 

103 if hasattr(self.tarfile, "extraction_filter") and hasattr( 

104 tarfile, "fully_trusted_filter" 

105 ): 

106 # Path and link safety checks happen in SafeTarFile before extraction. 

107 self.tarfile.extraction_filter = tarfile.fully_trusted_filter 

108 self.directories = {} 

109 

110 def close(self): 

111 self.tarfile.close() 

112 

113 def extractall(self, extract_root: Path): 

114 for member in self.tarfile.getmembers(): 

115 try: 

116 self.extract(member, extract_root) 

117 except Exception as e: 

118 self.record_problem(member, str(e), "Ignored.") 

119 self.fix_directories(extract_root) 

120 

121 def extract(self, tarinfo: tarfile.TarInfo, extract_root: Path): # noqa: C901 

122 if not tarinfo.name: 

123 self.record_problem( 

124 tarinfo, 

125 "File with empty filename in tar archive.", 

126 "Skipped.", 

127 ) 

128 return 

129 

130 if len(tarinfo.name) > MAX_PATH_LEN: 

131 self.record_problem( 

132 tarinfo, 

133 "File with filename too long in tar archive.", 

134 "Skipped.", 

135 ) 

136 return 

137 

138 if not RUNNING_AS_ROOT and (tarinfo.ischr() or tarinfo.isblk()): 

139 self.record_problem( 

140 tarinfo, 

141 "Missing elevated permissions for block and character device creation.", 

142 "Skipped.", 

143 ) 

144 return 

145 

146 # we do want to extract absolute paths, but they must be changed to prevent path traversal 

147 if Path(tarinfo.name).is_absolute(): 

148 self.record_problem( 

149 tarinfo, 

150 "Absolute path.", 

151 "Converted to extraction relative path.", 

152 ) 

153 tarinfo.name = str(Path(tarinfo.name).relative_to("/")) 

154 

155 # prevent traversal attempts through file name 

156 if not is_safe_path(basedir=extract_root, path=extract_root / tarinfo.name): 

157 self.record_problem( 

158 tarinfo, 

159 "Traversal attempt.", 

160 "Skipped.", 

161 ) 

162 return 

163 

164 # prevent traversal attempts through links 

165 if tarinfo.islnk() or tarinfo.issym(): 

166 if Path(tarinfo.linkname).is_absolute(): 

167 if tarinfo.islnk(): 

168 link_path = Path(tarinfo.linkname) 

169 relative_linkname = str(link_path.relative_to(link_path.anchor)) 

170 else: 

171 root = extract_root.resolve() 

172 path = (extract_root / tarinfo.name).resolve() 

173 

174 if path.parts[: len(root.parts)] != root.parts: 

175 self.record_problem( 

176 tarinfo, 

177 "Absolute path conversion to extraction relative failed - would escape root.", 

178 "Skipped.", 

179 ) 

180 return 

181 

182 depth = max(0, len(path.parts) - len(root.parts) - 1) 

183 relative_linkname = ( 

184 "/".join([".."] * depth) or "." 

185 ) + tarinfo.linkname 

186 

187 assert not Path(relative_linkname).is_absolute() 

188 self.record_problem( 

189 tarinfo, 

190 "Absolute path as link target.", 

191 "Converted to extraction relative path.", 

192 ) 

193 tarinfo.linkname = relative_linkname 

194 

195 # Symbolic link targets are relative to the link's parent, while tar 

196 # hard link targets are relative to the extraction root. 

197 if tarinfo.islnk(): 

198 resolved_path = extract_root / tarinfo.linkname 

199 else: 

200 resolved_path = (extract_root / tarinfo.name).parent / tarinfo.linkname 

201 if not is_safe_path(basedir=extract_root, path=resolved_path): 

202 self.record_problem( 

203 tarinfo, 

204 "Traversal attempt through link path.", 

205 "Skipped.", 

206 ) 

207 return 

208 

209 target_path = extract_root / tarinfo.name 

210 # directories are special: we can not set their metadata now + they might also be already existing 

211 if tarinfo.isdir(): 

212 # save (potentially duplicate) dir metadata for applying at the end of the extraction 

213 self.directories[tarinfo.name] = tarinfo 

214 target_path.mkdir(parents=True, exist_ok=True) 

215 return 

216 

217 if target_path.exists(): 

218 self.record_problem( 

219 tarinfo, 

220 "Duplicate tar entry.", 

221 "Removed older version.", 

222 ) 

223 target_path.unlink() 

224 

225 self.tarfile.extract(tarinfo, extract_root) 

226 

227 def fix_directories(self, extract_root): 

228 """Complete directory extraction. 

229 

230 When extracting directories, setting metadata was intentionally skipped, 

231 so that entries under the directory can be extracted, even if the directory 

232 is write protected. 

233 """ 

234 # need to set the permissions from leafs to root 

235 directories = sorted( 

236 self.directories.values(), key=lambda d: d.name, reverse=True 

237 ) 

238 

239 # copied from tarfile.extractall(), it is somewhat ugly, as uses private helpers! 

240 for tarinfo in directories: 

241 dirpath = str(extract_root / tarinfo.name) 

242 try: 

243 self.tarfile.chown(tarinfo, dirpath, numeric_owner=True) 

244 self.tarfile.utime(tarinfo, dirpath) 

245 self.tarfile.chmod(tarinfo, dirpath) 

246 except tarfile.ExtractError as e: 

247 self.record_problem(tarinfo, str(e), "Ignored.") 

248 

249 def record_problem(self, tarinfo, problem, resolution): 

250 logger.warning(f"{problem} {resolution}", path=tarinfo.name) # noqa: G004 

251 self.reports.append( 

252 ExtractionProblem( 

253 path=tarinfo.name, 

254 problem=problem, 

255 resolution=resolution, 

256 ) 

257 )