2 from bz2 import BZ2Decompressor
3 from zlib import decompressobj, MAX_WBITS
4 from gzip import FCOMMENT, FEXTRA, FHCRC, FNAME, FTEXT
5 from urlparse import urlparse
8 from twisted.python import log, filepath
9 from twisted.internet import defer
10 from twisted.trial import unittest
11 from twisted.web2 import stream
12 from twisted.web2.http import splitHostPort
14 from AptPackages import AptPackages
18 DECOMPRESS_EXTS = ['.gz', '.bz2']
19 DECOMPRESS_FILES = ['release', 'sources', 'packages']
21 class MirrorError(Exception):
22 """Exception raised when there's a problem with the mirror."""
24 class ProxyFileStream(stream.SimpleStream):
25 """Saves a stream to a file while providing a new stream."""
27 def __init__(self, stream, outFile, decompress = None, decFile = None):
28 """Initializes the proxy.
30 @type stream: C{twisted.web2.stream.IByteStream}
31 @param stream: the input stream to read from
32 @type outFile: C{twisted.python.filepath.FilePath}
33 @param outFile: the file to write to
34 @type decompress: C{string}
35 @param decompress: also decompress the file as this type
36 (currently only '.gz' and '.bz2' are supported)
37 @type decFile: C{twisted.python.filepath.FilePath}
38 @param decFile: the file to write the decompressed data to
41 self.outFile = outFile.open('w')
44 if decompress == ".gz":
46 self.gzfile = decFile.open('w')
47 self.gzdec = decompressobj(-MAX_WBITS)
48 elif decompress == ".bz2":
49 self.bz2file = decFile.open('w')
50 self.bz2dec = BZ2Decompressor()
51 self.length = self.stream.length
53 self.doneDefer = defer.Deferred()
56 """Close the output file."""
57 if not self.outFile.closed:
60 data_dec = self.gzdec.flush()
61 self.gzfile.write(data_dec)
68 self.doneDefer.callback(1)
71 """Read some data from the stream."""
72 if self.outFile.closed:
75 data = self.stream.read()
76 if isinstance(data, defer.Deferred):
77 data.addCallbacks(self._write, self._done)
83 def _write(self, data):
84 """Write the stream data to the file and return it for others to use."""
89 self.outFile.write(data)
93 new_data = self._remove_gzip_header(data)
94 dec_data = self.gzdec.decompress(new_data)
96 dec_data = self.gzdec.decompress(data)
97 self.gzfile.write(dec_data)
99 dec_data = self.bz2dec.decompress(data)
100 self.bz2file.write(dec_data)
103 def _remove_gzip_header(self, data):
104 if data[:2] != '\037\213':
105 raise IOError, 'Not a gzipped file'
106 if ord(data[2]) != 8:
107 raise IOError, 'Unknown compression method'
109 # modtime = self.fileobj.read(4)
110 # extraflag = self.fileobj.read(1)
111 # os = self.fileobj.read(1)
115 # Read & discard the extra field, if present
117 xlen = xlen + 256*ord(data[11])
118 skip = skip + 2 + xlen
120 # Read and discard a null-terminated string containing the filename
122 if not data[skip] or data[skip] == '\000':
127 # Read and discard a null-terminated string containing a comment
129 if not data[skip] or data[skip] == '\000':
134 skip += 2 # Read & discard the 16-bit header CRC
138 """Clean everything up and return None to future reads."""
144 """Manages all requests for mirror objects."""
146 def __init__(self, cache_dir):
147 self.cache_dir = cache_dir
148 self.cache = filepath.FilePath(self.cache_dir)
151 def extractPath(self, url):
152 parsed = urlparse(url)
153 host, port = splitHostPort(parsed[0], parsed[1])
154 site = host + ":" + str(port)
157 i = max(path.rfind('/dists/'), path.rfind('/pool/'))
162 # Uh oh, this is not good
163 log.msg("Couldn't find a good base directory for path: %s" % (site + path))
165 if site in self.apt_caches:
167 for base in self.apt_caches[site]:
169 for dirs in path.split('/'):
170 if base.startswith(base_match + '/' + dirs):
171 base_match += '/' + dirs
174 if len(base_match) > longest_match:
175 longest_match = len(base_match)
177 log.msg("Settled on baseDir: %s" % baseDir)
179 return site, baseDir, path
181 def init(self, site, baseDir):
182 if site not in self.apt_caches:
183 self.apt_caches[site] = {}
185 if baseDir not in self.apt_caches[site]:
186 site_cache = os.path.join(self.cache_dir, aptpkg_dir, 'mirrors', site + baseDir.replace('/', '_'))
187 self.apt_caches[site][baseDir] = AptPackages(site_cache)
189 def findHash(self, url):
190 site, baseDir, path = self.extractPath(url)
191 if site in self.apt_caches and baseDir in self.apt_caches[site]:
192 return self.apt_caches[site][baseDir].findHash(path)
194 d.errback(MirrorError("Site Not Found"))
197 def save_file(self, response, hash, size, url):
198 """Save a downloaded file to the cache and stream it."""
199 log.msg('Returning file: %s' % url)
201 parsed = urlparse(url)
202 destFile = self.cache.preauthChild(parsed[1] + parsed[2])
203 log.msg('Saving returned %r byte file to cache: %s' % (response.stream.length, destFile.path))
205 if destFile.exists():
206 log.msg('File already exists, removing: %s' % destFile.path)
209 destFile.parent().makedirs()
211 root, ext = os.path.splitext(destFile.basename())
212 if root.lower() in DECOMPRESS_FILES and ext.lower() in DECOMPRESS_EXTS:
214 decFile = destFile.sibling(root)
215 log.msg('Decompressing to: %s' % decFile.path)
217 log.msg('File already exists, removing: %s' % decFile.path)
223 orig_stream = response.stream
224 response.stream = ProxyFileStream(orig_stream, destFile, ext, decFile)
225 response.stream.doneDefer.addCallback(self.save_complete, url, destFile,
226 response.headers.getHeader('Last-Modified'),
228 response.stream.doneDefer.addErrback(self.save_error, url)
231 def save_complete(self, result, url, destFile, modtime = None, ext = None, decFile = None):
232 """Update the modification time and AptPackages."""
234 os.utime(destFile.path, (modtime, modtime))
236 os.utime(decFile.path, (modtime, modtime))
238 site, baseDir, path = self.extractPath(url)
239 self.init(site, baseDir)
240 self.apt_caches[site][baseDir].file_updated(path, destFile.path)
242 self.apt_caches[site][baseDir].file_updated(path[:-len(ext)], decFile.path)
244 def save_error(self, failure, url):
245 """An error has occurred in downloadign or saving the file."""
246 log.msg('Error occurred downloading %s' % url)
250 class TestMirrorManager(unittest.TestCase):
251 """Unit tests for the mirror manager."""
258 self.client = MirrorManager('/tmp')
260 def test_extractPath(self):
261 site, baseDir, path = self.client.extractPath('http://ftp.us.debian.org/debian/dists/unstable/Release')
262 self.failUnless(site == "ftp.us.debian.org:80", "no match: %s" % site)
263 self.failUnless(baseDir == "/debian", "no match: %s" % baseDir)
264 self.failUnless(path == "/dists/unstable/Release", "no match: %s" % path)
266 site, baseDir, path = self.client.extractPath('http://ftp.us.debian.org:16999/debian/pool/d/dpkg/dpkg_1.2.1-1.tar.gz')
267 self.failUnless(site == "ftp.us.debian.org:16999", "no match: %s" % site)
268 self.failUnless(baseDir == "/debian", "no match: %s" % baseDir)
269 self.failUnless(path == "/pool/d/dpkg/dpkg_1.2.1-1.tar.gz", "no match: %s" % path)
271 site, baseDir, path = self.client.extractPath('http://debian.camrdale.org/dists/unstable/Release')
272 self.failUnless(site == "debian.camrdale.org:80", "no match: %s" % site)
273 self.failUnless(baseDir == "", "no match: %s" % baseDir)
274 self.failUnless(path == "/dists/unstable/Release", "no match: %s" % path)
276 def verifyHash(self, found_hash, path, true_hash):
277 self.failUnless(found_hash[0] == true_hash,
278 "%s hashes don't match: %s != %s" % (path, found_hash[0], true_hash))
280 def test_findHash(self):
281 self.packagesFile = os.popen('ls -Sr /var/lib/apt/lists/ | grep -E "_main_.*Packages$" | tail -n 1').read().rstrip('\n')
282 self.sourcesFile = os.popen('ls -Sr /var/lib/apt/lists/ | grep -E "_main_.*Sources$" | tail -n 1').read().rstrip('\n')
283 for f in os.walk('/var/lib/apt/lists').next()[2]:
284 if f[-7:] == "Release" and self.packagesFile.startswith(f[:-7]):
288 self.client.updatedFile('http://' + self.releaseFile.replace('_','/'),
289 '/var/lib/apt/lists/' + self.releaseFile)
290 self.client.updatedFile('http://' + self.releaseFile[:self.releaseFile.find('_dists_')+1].replace('_','/') +
291 self.packagesFile[self.packagesFile.find('_dists_')+1:].replace('_','/'),
292 '/var/lib/apt/lists/' + self.packagesFile)
293 self.client.updatedFile('http://' + self.releaseFile[:self.releaseFile.find('_dists_')+1].replace('_','/') +
294 self.sourcesFile[self.sourcesFile.find('_dists_')+1:].replace('_','/'),
295 '/var/lib/apt/lists/' + self.sourcesFile)
297 lastDefer = defer.Deferred()
299 idx_hash = os.popen('grep -A 3000 -E "^SHA1:" ' +
300 '/var/lib/apt/lists/' + self.releaseFile +
301 ' | grep -E " main/binary-i386/Packages.bz2$"'
302 ' | head -n 1 | cut -d\ -f 2').read().rstrip('\n')
303 idx_path = 'http://' + self.releaseFile.replace('_','/')[:-7] + 'main/binary-i386/Packages.bz2'
305 d = self.client.findHash(idx_path)
306 d.addCallback(self.verifyHash, idx_path, idx_hash)
308 pkg_hash = os.popen('grep -A 30 -E "^Package: dpkg$" ' +
309 '/var/lib/apt/lists/' + self.packagesFile +
310 ' | grep -E "^SHA1:" | head -n 1' +
311 ' | cut -d\ -f 2').read().rstrip('\n')
312 pkg_path = 'http://' + self.releaseFile[:self.releaseFile.find('_dists_')+1].replace('_','/') + \
313 os.popen('grep -A 30 -E "^Package: dpkg$" ' +
314 '/var/lib/apt/lists/' + self.packagesFile +
315 ' | grep -E "^Filename:" | head -n 1' +
316 ' | cut -d\ -f 2').read().rstrip('\n')
318 d = self.client.findHash(pkg_path)
319 d.addCallback(self.verifyHash, pkg_path, pkg_hash)
321 src_dir = os.popen('grep -A 30 -E "^Package: dpkg$" ' +
322 '/var/lib/apt/lists/' + self.sourcesFile +
323 ' | grep -E "^Directory:" | head -n 1' +
324 ' | cut -d\ -f 2').read().rstrip('\n')
325 src_hashes = os.popen('grep -A 20 -E "^Package: dpkg$" ' +
326 '/var/lib/apt/lists/' + self.sourcesFile +
327 ' | grep -A 4 -E "^Files:" | grep -E "^ " ' +
328 ' | cut -d\ -f 2').read().split('\n')[:-1]
329 src_paths = os.popen('grep -A 20 -E "^Package: dpkg$" ' +
330 '/var/lib/apt/lists/' + self.sourcesFile +
331 ' | grep -A 4 -E "^Files:" | grep -E "^ " ' +
332 ' | cut -d\ -f 4').read().split('\n')[:-1]
334 for i in range(len(src_hashes)):
335 src_path = 'http://' + self.releaseFile[:self.releaseFile.find('_dists_')+1].replace('_','/') + src_dir + '/' + src_paths[i]
336 d = self.client.findHash(src_path)
337 d.addCallback(self.verifyHash, src_path, src_hashes[i])
339 idx_hash = os.popen('grep -A 3000 -E "^SHA1:" ' +
340 '/var/lib/apt/lists/' + self.releaseFile +
341 ' | grep -E " main/source/Sources.bz2$"'
342 ' | head -n 1 | cut -d\ -f 2').read().rstrip('\n')
343 idx_path = 'http://' + self.releaseFile.replace('_','/')[:-7] + 'main/source/Sources.bz2'
345 d = self.client.findHash(idx_path)
346 d.addCallback(self.verifyHash, idx_path, idx_hash)
348 d.addBoth(lastDefer.callback)
352 for p in self.pending_calls: