Coverage for daklib/import_repository.py: 26%

1# Copyright (C) 2015, Ansgar Burchardt <ansgar@debian.org> 

2# 

3# This program is free software; you can redistribute it and/or modify 

4# it under the terms of the GNU General Public License as published by 

5# the Free Software Foundation; either version 2 of the License, or 

6# (at your option) any later version. 

7# 

8# This program is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU General Public License for more details. 

12# 

13# You should have received a copy of the GNU General Public License along 

14# with this program; if not, write to the Free Software Foundation, Inc., 

15# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 

16 

17# This is still work-in-progress and far too incomplete. 

18# ruff: noqa 

19# type: ignore 

20 

21import os 

22import shutil 

23import tempfile 

24import urllib.error 

25import urllib.parse 

26import urllib.request 

27from typing import Optional 

28 

29import apt_pkg 

30from sqlalchemy import select 

31from sqlalchemy.orm import object_session 

32 

33import daklib.compress 

34import daklib.config 

35import daklib.dakapt 

36import daklib.dbconn 

37import daklib.gpg 

38import daklib.regexes 

39import daklib.upload 

40from daklib.dbconn import Archive, DBBinary, DBSource, PoolFile 

41 

42# Hmm, maybe use APT directly for all of this? 

43 

44_release_hashes_fields = ("MD5Sum", "SHA1", "SHA256") 

45 

46 

47class Release: 

48 def __init__(self, base, suite_name, data): 

49 self._base = base 

50 self._suite_name = suite_name 

51 self._dict = apt_pkg.TagSection(data) 

52 self._hashes = daklib.upload.parse_file_list( 

53 self._dict, False, daklib.regexes.re_file_safe_slash, _release_hashes_fields 

54 ) 

55 

56 def architectures(self): 

57 return self._dict["Architectures"].split() 

58 

59 def components(self): 

60 return self._dict["Components"].split() 

61 

62 def packages(self, component, architecture): 

63 fn = "{0}/binary-{1}/Packages".format(component, architecture) 

64 tmp = obtain_release_file(self, fn) 

65 return apt_pkg.TagFile(tmp.fh()) 

66 

67 def sources(self, component): 

68 fn = "{0}/source/Sources".format(component) 

69 tmp = obtain_release_file(self, fn) 

70 return apt_pkg.TagFile(tmp.fh()) 

71 

72 def suite(self): 

73 return self._dict["Suite"] 

74 

75 def codename(self): 

76 return self._dict["Codename"] 

77 

78 # TODO: Handle Date/Valid-Until to make sure we import 

79 # a newer version than before 

80 

81 

82class File: 

83 def __init__(self): 

84 config = daklib.config.Config() 

85 self._tmp = tempfile.NamedTemporaryFile(dir=config["Dir::TempPath"]) 

86 

87 def fh(self): 

88 self._tmp.seek(0) 

89 return self._tmp 

90 

91 def hashes(self): 

92 return daklib.dakapt.DakHashes(self.fh()) 

93 

94 

95def obtain_file(base, path) -> File: 

96 """Obtain a file 'path' located below 'base' 

97 

98 .. note:: 

99 

100 return type can still change 

101 """ 

102 fn = "{0}/{1}".format(base, path) 

103 tmp = File() 

104 if fn.startswith("http://"): 

105 fh = urllib.request.urlopen(fn, timeout=300) 

106 shutil.copyfileobj(fh, tmp._tmp) 

107 fh.close() 

108 else: 

109 with open(fn, "rb") as fh: 

110 shutil.copyfileobj(fh, tmp._tmp) 

111 return tmp 

112 

113 

114def obtain_release(base, suite_name, keyring, fingerprint=None) -> Release: 

115 """Obtain release information""" 

116 tmp = obtain_file(base, "dists/{0}/InRelease".format(suite_name)) 

117 data = tmp.fh().read() 

118 f = daklib.gpg.SignedFile(data, [keyring]) 

119 r = Release(base, suite_name, f.contents) 

120 if r.suite() != suite_name and r.codename() != suite_name: 

121 raise Exception( 

122 "Suite {0} doesn't match suite or codename from Release file.".format( 

123 suite_name 

124 ) 

125 ) 

126 return r 

127 

128 

129_compressions = (".zst", ".xz", ".gz", ".bz2") 

130 

131 

132def obtain_release_file(release, filename) -> File: 

133 """Obtain file referenced from Release 

134 

135 A compressed version is automatically selected and decompressed if it exists. 

136 """ 

137 if filename not in release._hashes: 

138 raise ValueError("File {0} not referenced in Release".format(filename)) 

139 

140 compressed = False 

141 for ext in _compressions: 

142 compressed_file = filename + ext 

143 if compressed_file in release._hashes: 

144 compressed = True 

145 filename = compressed_file 

146 break 

147 

148 # Obtain file and check hashes 

149 tmp = obtain_file( 

150 release._base, "dists/{0}/{1}".format(release._suite_name, filename) 

151 ) 

152 hashedfile = release._hashes[filename] 

153 hashedfile.check_fh(tmp.fh()) 

154 

155 if compressed: 

156 tmp2 = File() 

157 daklib.compress.decompress(tmp.fh(), tmp2.fh(), filename) 

158 tmp = tmp2 

159 

160 return tmp 

161 

162 

163def import_source_to_archive(base, entry, transaction, archive, component) -> DBSource: 

164 """Import source package described by 'entry' into the given 'archive' and 'component' 

165 

166 'entry' needs to be a dict-like object with at least the following 

167 keys as used in a Sources index: Directory, Files, Checksums-Sha1, 

168 Checksums-Sha256 

169 """ 

170 # Obtain and verify files 

171 if not daklib.regexes.re_file_safe_slash.match(entry["Directory"]): 

172 raise Exception("Unsafe path in Directory field") 

173 hashed_files = daklib.upload.parse_file_list(entry, False) 

174 files = [] 

175 for f in hashed_files.values(): 

176 path = os.path.join(entry["Directory"], f.filename) 

177 tmp = obtain_file(base, path) 

178 f.check_fh(tmp.fh()) 

179 files.append(tmp) 

180 directory, f.input_filename = os.path.split(tmp.fh().name) 

181 

182 # Inject files into archive 

183 source = daklib.upload.Source( 

184 directory, list(hashed_files.values()), [], require_signature=False 

185 ) 

186 # TODO: ugly hack! 

187 for f in hashed_files.keys(): 

188 if f.endswith(".dsc"): 

189 continue 

190 source.files[f].input_filename = hashed_files[f].input_filename 

191 

192 # TODO: allow changed_by to be NULL 

193 changed_by = source.dsc["Maintainer"] 

194 db_changed_by = daklib.dbconn.get_or_set_maintainer(changed_by, transaction.session) 

195 db_source = transaction.install_source_to_archive( 

196 directory, source, archive, component, db_changed_by 

197 ) 

198 

199 return db_source 

200 

201 

202def import_package_to_suite(base, entry, transaction, suite, component) -> DBBinary: 

203 """Import binary package described by 'entry' into the given 'suite' and 'component' 

204 

205 'entry' needs to be a dict-like object with at least the following 

206 keys as used in a Packages index: Filename, Size, MD5sum, SHA1, 

207 SHA256 

208 """ 

209 # Obtain and verify file 

210 filename = entry["Filename"] 

211 tmp = obtain_file(base, filename) 

212 directory, fn = os.path.split(tmp.fh().name) 

213 hashedfile = daklib.upload.HashedFile( 

214 os.path.basename(filename), 

215 int(entry["Size"]), 

216 entry["MD5sum"], 

217 entry["SHA1"], 

218 entry["SHA256"], 

219 input_filename=fn, 

220 ) 

221 hashedfile.check_fh(tmp.fh()) 

222 

223 # Inject file into archive 

224 binary = daklib.upload.Binary(directory, hashedfile) 

225 db_binary = transaction.install_binary(directory, binary, suite, component) 

226 transaction.flush() 

227 

228 return db_binary 

229 

230 

231def import_source_to_suite(base, entry, transaction, suite, component): 

232 """Import source package described by 'entry' into the given 'suite' and 'component' 

233 

234 'entry' needs to be a dict-like object with at least the following 

235 keys as used in a Sources index: Directory, Files, Checksums-Sha1, 

236 Checksums-Sha256 

237 """ 

238 source = import_source_to_archive( 

239 base, entry, transaction, suite.archive, component 

240 ) 

241 source.suites.append(suite) 

242 transaction.flush() 

243 

244 

245def source_in_archive( 

246 source: str, 

247 version: str, 

248 archive: Archive, 

249 component: Optional[daklib.dbconn.Component] = None, 

250) -> bool: 

251 """Check that source package 'source' with version 'version' exists in 'archive', 

252 with an optional check for the given component 'component'. 

253 

254 .. note:: 

255 

256 This should probably be moved somewhere else 

257 """ 

258 session = object_session(archive) 

259 query = ( 

260 select(DBSource) 

261 .filter_by(source=source, version=version) 

262 .join(DBSource.poolfile) 

263 .join(PoolFile.archives) 

264 .filter_by(archive=archive) 

265 ) 

266 if component is not None: 

267 query = query.filter_by(component=component) 

268 return bool(session.scalar(select(query.exists())))