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_knownchange
  36
  37 from daklib.config import Config
  38
  39 # where in dak.conf all of our configuration will be stowed
  40 options_prefix = "KnownChanges"
  41 options_prefix = "%s::Options" % options_prefix
  42
  43 log = logging.getLogger()
  44
  45 ################################################################################
  46
  47
  48 def usage (exit_code=0):
  49     print """Usage: dak import-known-changes [options]
  50
  51 OPTIONS
  52      -j n
  53         run with n threads concurrently
  54
  55      -v, --verbose
  56         show verbose information messages
  57
  58      -q, --quiet
  59         supress all output but errors
  60
  61 """
  62     sys.exit(exit_code)
  63
  64 def check_signature (sig_filename, data_filename=""):
  65     keyrings = [
  66         "/home/joerg/keyring/keyrings/debian-keyring.gpg",
  67         "/home/joerg/keyring/keyrings/debian-keyring.pgp",
  68         "/home/joerg/keyring/keyrings/debian-maintainers.gpg",
  69         "/home/joerg/keyring/keyrings/debian-role-keys.gpg",
  70         "/home/joerg/keyring/keyrings/emeritus-keyring.pgp",
  71         "/home/joerg/keyring/keyrings/emeritus-keyring.gpg",
  72         "/home/joerg/keyring/keyrings/removed-keys.gpg",
  73         "/home/joerg/keyring/keyrings/removed-keys.pgp"
  74         ]
  75
  76     keyringargs = " ".join(["--keyring %s" % x for x in keyrings ])
  77
  78     # Build the command line
  79     status_read, status_write = os.pipe()
  80     cmd = "gpgv --status-fd %s %s %s" % (status_write, keyringargs, sig_filename)
  81
  82     # Invoke gpgv on the file
  83     (output, status, exit_status) = gpgv_get_status_output(cmd, status_read, status_write)
  84
  85     # Process the status-fd output
  86     (keywords, internal_error) = process_gpgv_output(status)
  87
  88     # If we failed to parse the status-fd output, let's just whine and bail now
  89     if internal_error:
  90         warn("Couldn't parse signature")
  91         return None
  92
  93     # usually one would check for bad things here. We, however, do not care.
  94
  95     # Next check gpgv exited with a zero return code
  96     if exit_status:
  97         warn("Couldn't parse signature")
  98         return None
  99
 100     # Sanity check the good stuff we expect
 101     if not keywords.has_key("VALIDSIG"):
 102         warn("Couldn't parse signature")
 103     else:
 104         args = keywords["VALIDSIG"]
 105         if len(args) < 1:
 106             warn("Couldn't parse signature")
 107         else:
 108             fingerprint = args[0]
 109
 110     return fingerprint
 111
 112
 113 class EndOfChanges(object):
 114     """something enqueued to signify the last change"""
 115     pass
 116
 117
 118 class OneAtATime(object):
 119     """
 120     a one space queue which sits between multiple possible producers
 121     and multiple possible consumers
 122     """
 123     def __init__(self):
 124         self.next_in_line = None
 125         self.next_lock = threading.Condition()
 126
 127     def enqueue(self, next):
 128         self.next_lock.acquire()
 129         while self.next_in_line:
 130             self.next_lock.wait()
 131
 132         assert( not self.next_in_line )
 133         self.next_in_line = next
 134         self.next_lock.notify()
 135         self.next_lock.release()
 136
 137     def dequeue(self):
 138         self.next_lock.acquire()
 139         while not self.next_in_line:
 140             self.next_lock.wait()
 141         result = self.next_in_line
 142
 143         if isinstance(next, EndOfChanges):
 144             return None
 145
 146         self.next_in_line = None
 147         self.next_lock.notify()
 148         self.next_lock.release()
 149         return result
 150
 151 class ChangesToImport(object):
 152     """A changes file to be enqueued to be processed"""
 153     def __init__(self, queue, checkdir, changesfile, count):
 154         self.queue = queue
 155         self.checkdir = checkdir
 156         self.changesfile = changesfile
 157         self.count = count
 158
 159 class ChangesGenerator(threading.Thread):
 160     """enqueues changes files to be imported"""
 161     def __init__(self, queue):
 162         self.queue = queue
 163         self.session = DBConn().session()
 164
 165     def run(self):
 166         cnf = Config()
 167         for directory in [ "Accepted", "Byhand", "Done", "New", "ProposedUpdates", "OldProposedUpdates" ]:
 168             checkdir = cnf["Dir::Queue::%s" % (directory) ]
 169             if os.path.exists(checkdir):
 170                 print "Looking into %s" % (checkdir)
 171
 172                 for dirpath, dirnames, filenames in os.walk(checkdir, topdown=False):
 173                     if not filenames:
 174                         # Empty directory (or only subdirectories), next
 175                         continue
 176                     for changesfile in filenames:
 177                         if not changesfile.endswith(".changes"):
 178                             # Only interested in changes files.
 179                             continue
 180                             count += 1
 181
 182                             if not get_knownchange(session):
 183                                 self.queue.enqueue(ChangesToImport(directory, checkdir, changesfile, count))
 184
 185         self.queue.enqueue(EndOfChanges())
 186
 187 class ImportThread(threading.Thread):
 188     def __init__(self, queue):
 189         self.queue = queue
 190         self.session = DBConn().session()
 191
 192     def run(self):
 193         while True:
 194             try:
 195                 to_import = queue.dequeue()
 196                 if not to_import:
 197                     return
 198
 199                 print( "Directory %s, file %7d, failures %3d. (%s)" % (to_import.dirpath[-10:], to_import.count, failure, to_import.changesfile) )
 200
 201                 changes = Changes()
 202                 changes.changes_file = to_import.changesfile
 203                 changesfile = os.path.join(to_import.dirpath, to_import.changesfile)
 204                 changes.changes = parse_changes(changesfile, signing_rules=-1)
 205                 changes.changes["fingerprint"] = check_signature(changesfile)
 206                 changes.add_known_changes(to_import.queue, self.session)
 207                 self.session.commit()
 208
 209             except InvalidDscError, line:
 210                 warn("syntax error in .dsc file '%s', line %s." % (f, line))
 211                 failure += 1
 212
 213             except ChangesUnicodeError:
 214                 warn("found invalid changes file, not properly utf-8 encoded")
 215                 failure += 1
 216
 217                 print "Directory %s, file %7d, failures %3d. (%s)" % (dirpath[-10:], count, failure, changesfile)
 218
 219
 220
 221 def main():
 222     cnf = Config()
 223
 224     arguments = [('h',"help", "%s::%s" % (options_prefix,"Help")),
 225                  ('j',"concurrency", "%s::%s" % (options_prefix,"Concurrency"),"HasArg"),
 226                  ('q',"quiet", "%s::%s" % (options_prefix,"Quiet")),
 227                  ('v',"verbose", "%s::%s" % (options_prefix,"Verbose")),
 228                 ]
 229
 230     args = apt_pkg.ParseCommandLine(cnf.Cnf, arguments,sys.argv)
 231
 232     num_threads = 1
 233
 234     if (len(args) < 1) or not commands.has_key(args[0]):
 235         usage()
 236
 237     if cnf.has_key("%s::%s" % (options_prefix,"Help")):
 238         usage()
 239
 240     level=logging.INFO
 241     if cnf.has_key("%s::%s" % (options_prefix,"Quiet")):
 242         level=logging.ERROR
 243
 244     elif cnf.has_key("%s::%s" % (options_prefix,"Verbose")):
 245         level=logging.DEBUG
 246
 247
 248     logging.basicConfig( level=level,
 249                          format='%(asctime)s %(levelname)s %(message)s',
 250                          stream = sys.stderr )
 251
 252     if Config().has_key( "%s::%s" %(options_prefix,"Concurrency")):
 253         num_threads = int(Config()[ "%s::%s" %(options_prefix,"Suite")])
 254
 255
 256     queue = OneAtATime()
 257     ChangesGenerator(queue).start()
 258
 259     for i in range(num_threads):
 260         ImportThread(queue).start()
 261
 262 def which_suites(session):
 263     """
 264     return a list of suites to operate on
 265     """
 266     if Config().has_key( "%s::%s" %(options_prefix,"Suite")):
 267         suites = utils.split_args(Config()[ "%s::%s" %(options_prefix,"Suite")])
 268     else:
 269         suites = Config().SubTree("Suite").List()
 270
 271     return [get_suite(s.lower(), session) for s in suites]
 272
 273
 274 if __name__ == '__main__':
 275     main()