]> git.decadent.org.uk Git - dak.git/blob - dak/import_known_changes.py
dinstall
[dak.git] / dak / import_known_changes.py
1 #!/usr/bin/env python
2 # coding=utf8
3
4 """
5 Import known_changes files
6
7 @contact: Debian FTP Master <ftpmaster@debian.org>
8 @copyright: 2009  Mike O'Connor <stew@debian.org>
9 @license: GNU General Public License version 2 or later
10 """
11
12 # This program is free software; you can redistribute it and/or modify
13 # it under the terms of the GNU General Public License as published by
14 # the Free Software Foundation; either version 2 of the License, or
15 # (at your option) any later version.
16
17 # This program is distributed in the hope that it will be useful,
18 # but WITHOUT ANY WARRANTY; without even the implied warranty of
19 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20 # GNU General Public License for more details.
21
22 # You should have received a copy of the GNU General Public License
23 # along with this program; if not, write to the Free Software
24 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
25
26 ################################################################################
27
28
29 ################################################################################
30
31 import sys
32 import os
33 import logging
34 import threading
35 from daklib.dbconn import DBConn, get_dbchange
36 from daklib.config import Config
37 import apt_pkg
38 from daklib.dak_exceptions import DBUpdateError, InvalidDscError, ChangesUnicodeError
39 from daklib.changes import Changes
40 from daklib.utils import parse_changes, warn, gpgv_get_status_output, process_gpgv_output
41 import traceback
42
43 # where in dak.conf all of our configuration will be stowed
44 options_prefix = "KnownChanges"
45 options_prefix = "%s::Options" % options_prefix
46
47 log = logging.getLogger()
48
49 ################################################################################
50
51
52 def usage (exit_code=0):
53     print """Usage: dak import-known-changes [options]
54
55 OPTIONS
56      -j n
57         run with n threads concurrently
58
59      -v, --verbose
60         show verbose information messages
61
62      -q, --quiet
63         supress all output but errors
64
65 """
66     sys.exit(exit_code)
67
68 def check_signature (sig_filename, data_filename=""):
69     fingerprint = None
70
71     keyrings = [
72         "/home/joerg/keyring/keyrings/debian-keyring.gpg",
73         "/home/joerg/keyring/keyrings/debian-keyring.pgp",
74         "/home/joerg/keyring/keyrings/debian-maintainers.gpg",
75         "/home/joerg/keyring/keyrings/debian-role-keys.gpg",
76         "/home/joerg/keyring/keyrings/emeritus-keyring.pgp",
77         "/home/joerg/keyring/keyrings/emeritus-keyring.gpg",
78         "/home/joerg/keyring/keyrings/removed-keys.gpg",
79         "/home/joerg/keyring/keyrings/removed-keys.pgp"
80         ]
81
82     keyringargs = " ".join(["--keyring %s" % x for x in keyrings ])
83
84     # Build the command line
85     status_read, status_write = os.pipe()
86     cmd = "gpgv --status-fd %s %s %s" % (status_write, keyringargs, sig_filename)
87
88     # Invoke gpgv on the file
89     (output, status, exit_status) = gpgv_get_status_output(cmd, status_read, status_write)
90
91     # Process the status-fd output
92     (keywords, internal_error) = process_gpgv_output(status)
93
94     # If we failed to parse the status-fd output, let's just whine and bail now
95     if internal_error:
96         warn("Couldn't parse signature")
97         return None
98
99     # usually one would check for bad things here. We, however, do not care.
100
101     # Next check gpgv exited with a zero return code
102     if exit_status:
103         warn("Couldn't parse signature")
104         return None
105
106     # Sanity check the good stuff we expect
107     if not keywords.has_key("VALIDSIG"):
108         warn("Couldn't parse signature")
109     else:
110         args = keywords["VALIDSIG"]
111         if len(args) < 1:
112             warn("Couldn't parse signature")
113         else:
114             fingerprint = args[0]
115
116     return fingerprint
117
118
119 class EndOfChanges(object):
120     """something enqueued to signify the last change"""
121     pass
122
123
124 class OneAtATime(object):
125     """
126     a one space queue which sits between multiple possible producers
127     and multiple possible consumers
128     """
129     def __init__(self):
130         self.next_in_line = None
131         self.read_lock = threading.Condition()
132         self.write_lock = threading.Condition()
133         self.die = False
134
135     def plsDie(self):
136         self.die = True
137         self.write_lock.acquire()
138         self.write_lock.notifyAll()
139         self.write_lock.release()
140
141         self.read_lock.acquire()
142         self.read_lock.notifyAll()
143         self.read_lock.release()
144
145     def enqueue(self, next):
146         self.write_lock.acquire()
147         while self.next_in_line:
148             if self.die:
149                 return
150             self.write_lock.wait()
151
152         assert( not self.next_in_line )
153         self.next_in_line = next
154         self.write_lock.release()
155         self.read_lock.acquire()
156         self.read_lock.notify()
157         self.read_lock.release()
158
159     def dequeue(self):
160         self.read_lock.acquire()
161         while not self.next_in_line:
162             if self.die:
163                 return
164             self.read_lock.wait()
165
166         result = self.next_in_line
167
168         self.next_in_line = None
169         self.read_lock.release()
170         self.write_lock.acquire()
171         self.write_lock.notify()
172         self.write_lock.release()
173
174         if isinstance(result, EndOfChanges):
175             return None
176
177         return result
178
179 class ChangesToImport(object):
180     """A changes file to be enqueued to be processed"""
181     def __init__(self, checkdir, changesfile, count):
182         self.dirpath = checkdir
183         self.changesfile = changesfile
184         self.count = count
185
186     def __str__(self):
187         return "#%d: %s in %s" % (self.count, self.changesfile, self.dirpath)
188
189 class ChangesGenerator(threading.Thread):
190     """enqueues changes files to be imported"""
191     def __init__(self, parent, queue):
192         threading.Thread.__init__(self)
193         self.queue = queue
194         self.session = DBConn().session()
195         self.parent = parent
196         self.die = False
197
198     def plsDie(self):
199         self.die = True
200
201     def run(self):
202         cnf = Config()
203         count = 1
204         for directory in [ "Accepted", "Byhand", "Done", "New", "ProposedUpdates", "OldProposedUpdates" ]:
205             checkdir = cnf["Dir::Queue::%s" % (directory) ]
206             if os.path.exists(checkdir):
207                 print "Looking into %s" % (checkdir)
208
209                 for dirpath, dirnames, filenames in os.walk(checkdir, topdown=True):
210                     if not filenames:
211                         # Empty directory (or only subdirectories), next
212                         continue
213
214                     for changesfile in filenames:
215                         try:
216                             if not changesfile.endswith(".changes"):
217                                 # Only interested in changes files.
218                                 continue
219                             count += 1
220
221                             if not get_dbchange(changesfile, self.session):
222                                 to_import = ChangesToImport(dirpath, changesfile, count)
223                                 if self.die:
224                                     return
225                                 self.queue.enqueue(to_import)
226                         except KeyboardInterrupt:
227                             print("got Ctrl-c in enqueue thread.  terminating")
228                             self.parent.plsDie()
229                             sys.exit(1)
230
231         self.queue.enqueue(EndOfChanges())
232
233 class ImportThread(threading.Thread):
234     def __init__(self, parent, queue):
235         threading.Thread.__init__(self)
236         self.queue = queue
237         self.session = DBConn().session()
238         self.parent = parent
239         self.die = False
240
241     def plsDie(self):
242         self.die = True
243
244     def run(self):
245         while True:
246             try:
247                 if self.die:
248                     return
249                 to_import = self.queue.dequeue()
250                 if not to_import:
251                     return
252
253                 print( "Directory %s, file %7d, (%s)" % (to_import.dirpath[-10:], to_import.count, to_import.changesfile) )
254
255                 changes = Changes()
256                 changes.changes_file = to_import.changesfile
257                 changesfile = os.path.join(to_import.dirpath, to_import.changesfile)
258                 changes.changes = parse_changes(changesfile, signing_rules=-1)
259                 changes.changes["fingerprint"] = check_signature(changesfile)
260                 changes.add_known_changes(to_import.dirpath, self.session)
261                 self.session.commit()
262
263             except InvalidDscError, line:
264                 warn("syntax error in .dsc file '%s', line %s." % (f, line))
265
266             except ChangesUnicodeError:
267                 warn("found invalid changes file, not properly utf-8 encoded")
268
269             except KeyboardInterrupt:
270                 print("Caught C-c; on ImportThread. terminating.")
271                 self.parent.plsDie()
272                 sys.exit(1)
273
274             except:
275                 self.parent.plsDie()
276                 sys.exit(1)
277
278 class ImportKnownChanges(object):
279     def __init__(self,num_threads):
280         self.queue = OneAtATime()
281         self.threads = [ ChangesGenerator(self,self.queue) ]
282
283         for i in range(num_threads):
284             self.threads.append( ImportThread(self,self.queue) )
285
286         try:
287             for thread in self.threads:
288                 thread.start()
289
290         except KeyboardInterrupt:
291             print("Caught C-c; terminating.")
292             utils.warn("Caught C-c; terminating.")
293             self.plsDie()
294
295     def plsDie(self):
296         traceback.print_stack90
297         for thread in self.threads:
298             print( "STU: before ask %s to die" % thread )
299             thread.plsDie()
300             print( "STU: after ask %s to die" % thread )
301
302         self.threads=[]
303         sys.exit(1)
304
305
306 def main():
307     cnf = Config()
308
309     arguments = [('h',"help", "%s::%s" % (options_prefix,"Help")),
310                  ('j',"concurrency", "%s::%s" % (options_prefix,"Concurrency"),"HasArg"),
311                  ('q',"quiet", "%s::%s" % (options_prefix,"Quiet")),
312                  ('v',"verbose", "%s::%s" % (options_prefix,"Verbose")),
313                 ]
314
315     args = apt_pkg.ParseCommandLine(cnf.Cnf, arguments,sys.argv)
316
317     num_threads = 1
318
319     if len(args) > 0:
320         usage()
321
322     if cnf.has_key("%s::%s" % (options_prefix,"Help")):
323         usage()
324
325     level=logging.INFO
326     if cnf.has_key("%s::%s" % (options_prefix,"Quiet")):
327         level=logging.ERROR
328
329     elif cnf.has_key("%s::%s" % (options_prefix,"Verbose")):
330         level=logging.DEBUG
331
332
333     logging.basicConfig( level=level,
334                          format='%(asctime)s %(levelname)s %(message)s',
335                          stream = sys.stderr )
336
337     if Config().has_key( "%s::%s" %(options_prefix,"Concurrency")):
338         num_threads = int(Config()[ "%s::%s" %(options_prefix,"Concurrency")])
339
340     ImportKnownChanges(num_threads)
341
342
343
344
345 if __name__ == '__main__':
346     main()