dak/process_policy.py

   1 #!/usr/bin/env python
   2 # vim:set et ts=4 sw=4:
   3
   4 """ Handles packages from policy queues
   5
   6 @contact: Debian FTP Master <ftpmaster@debian.org>
   7 @copyright: 2001, 2002, 2003, 2004, 2005, 2006  James Troup <james@nocrew.org>
   8 @copyright: 2009 Joerg Jaspert <joerg@debian.org>
   9 @copyright: 2009 Frank Lichtenheld <djpig@debian.org>
  10 @copyright: 2009 Mark Hymers <mhy@debian.org>
  11 @license: GNU General Public License version 2 or later
  12 """
  13 # This program is free software; you can redistribute it and/or modify
  14 # it under the terms of the GNU General Public License as published by
  15 # the Free Software Foundation; either version 2 of the License, or
  16 # (at your option) any later version.
  17
  18 # This program is distributed in the hope that it will be useful,
  19 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  20 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  21 # GNU General Public License for more details.
  22
  23 # You should have received a copy of the GNU General Public License
  24 # along with this program; if not, write to the Free Software
  25 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  26
  27 ################################################################################
  28
  29 # <mhy> So how do we handle that at the moment?
  30 # <stew> Probably incorrectly.
  31
  32 ################################################################################
  33
  34 import os
  35 import datetime
  36 import re
  37 import sys
  38 import traceback
  39 import apt_pkg
  40
  41 from daklib.dbconn import *
  42 from daklib import daklog
  43 from daklib import utils
  44 from daklib.dak_exceptions import CantOpenError, AlreadyLockedError, CantGetLockError
  45 from daklib.config import Config
  46 from daklib.archive import ArchiveTransaction
  47 from daklib.urgencylog import UrgencyLog
  48 from daklib.textutils import fix_maintainer
  49
  50 # Globals
  51 Options = None
  52 Logger = None
  53
  54 ################################################################################
  55
  56 def do_comments(dir, srcqueue, opref, npref, line, fn, transaction):
  57     session = transaction.session
  58     for comm in [ x for x in os.listdir(dir) if x.startswith(opref) ]:
  59         lines = open(os.path.join(dir, comm)).readlines()
  60         if len(lines) == 0 or lines[0] != line + "\n": continue
  61
  62         # If the ACCEPT includes a _<arch> we only accept that .changes.
  63         # Otherwise we accept all .changes that start with the given prefix
  64         changes_prefix = comm[len(opref):]
  65         if changes_prefix.count('_') < 2:
  66             changes_prefix = changes_prefix + '_'
  67         else:
  68             changes_prefix = changes_prefix + '.changes'
  69
  70         uploads = session.query(PolicyQueueUpload).filter_by(policy_queue=srcqueue) \
  71             .join(PolicyQueueUpload.changes).filter(DBChange.changesname.startswith(changes_prefix)) \
  72             .order_by(PolicyQueueUpload.source_id)
  73         for u in uploads:
  74             print "Processing changes file: %s" % u.changes.changesname
  75             fn(u, srcqueue, "".join(lines[1:]), transaction)
  76
  77         if opref != npref:
  78             newcomm = npref + comm[len(opref):]
  79             transaction.fs.move(os.path.join(dir, comm), os.path.join(dir, newcomm))
  80
  81 ################################################################################
  82
  83 def try_or_reject(function):
  84     def wrapper(upload, srcqueue, comments, transaction):
  85         try:
  86             function(upload, srcqueue, comments, transaction)
  87         except Exception as e:
  88             comments = 'An exception was raised while processing the package:\n{0}\nOriginal comments:\n{1}'.format(traceback.format_exc(), comments)
  89             try:
  90                 transaction.rollback()
  91                 real_comment_reject(upload, srcqueue, comments, transaction)
  92             except Exception as e:
  93                 comments = 'In addition an exception was raised while trying to reject the upload:\n{0}\nOriginal rejection:\n{1}'.format(traceback.format_exc(), comments)
  94                 transaction.rollback()
  95                 real_comment_reject(upload, srcqueue, comments, transaction, notify=False)
  96         if not Options['No-Action']:
  97             transaction.commit()
  98     return wrapper
  99
 100 ################################################################################
 101
 102 @try_or_reject
 103 def comment_accept(upload, srcqueue, comments, transaction):
 104     for byhand in upload.byhand:
 105         path = os.path.join(srcqueue.path, byhand.filename)
 106         if os.path.exists(path):
 107             raise Exception('E: cannot ACCEPT upload with unprocessed byhand file {0}'.format(byhand.filename))
 108
 109     cnf = Config()
 110
 111     fs = transaction.fs
 112     session = transaction.session
 113     changesname = upload.changes.changesname
 114     allow_tainted = srcqueue.suite.archive.tainted
 115
 116     # We need overrides to get the target component
 117     overridesuite = upload.target_suite
 118     if overridesuite.overridesuite is not None:
 119         overridesuite = session.query(Suite).filter_by(suite_name=overridesuite.overridesuite).one()
 120
 121     def binary_component_func(db_binary):
 122         override = session.query(Override).filter_by(suite=overridesuite, package=db_binary.package) \
 123             .join(OverrideType).filter(OverrideType.overridetype == db_binary.binarytype) \
 124             .join(Component).one()
 125         return override.component
 126
 127     def source_component_func(db_source):
 128         override = session.query(Override).filter_by(suite=overridesuite, package=db_source.source) \
 129             .join(OverrideType).filter(OverrideType.overridetype == 'dsc') \
 130             .join(Component).one()
 131         return override.component
 132
 133     all_target_suites = [upload.target_suite]
 134     all_target_suites.extend([q.suite for q in upload.target_suite.copy_queues])
 135
 136     for suite in all_target_suites:
 137         if upload.source is not None:
 138             transaction.copy_source(upload.source, suite, source_component_func(upload.source), allow_tainted=allow_tainted)
 139         for db_binary in upload.binaries:
 140             transaction.copy_binary(db_binary, suite, binary_component_func(db_binary), allow_tainted=allow_tainted, extra_archives=[upload.target_suite.archive])
 141
 142     # Copy .changes if needed
 143     if upload.target_suite.copychanges:
 144         src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 145         dst = os.path.join(upload.target_suite.path, upload.changes.changesname)
 146         fs.copy(src, dst, mode=upload.target_suite.archive.mode)
 147
 148     if upload.source is not None and not Options['No-Action']:
 149         urgency = upload.changes.urgency
 150         if urgency not in cnf.value_list('Urgency::Valid'):
 151             urgency = cnf['Urgency::Default']
 152         UrgencyLog().log(upload.source.source, upload.source.version, urgency)
 153
 154     print "  ACCEPT"
 155     if not Options['No-Action']:
 156         Logger.log(["Policy Queue ACCEPT", srcqueue.queue_name, changesname])
 157
 158     # Send announcement
 159     if upload.source is not None:
 160         subst = subst_for_upload(upload)
 161         announce = ", ".join(upload.target_suite.announce or [])
 162         tracking = cnf.get('Dinstall::TrackingServer')
 163         if tracking and upload.source is not None:
 164             announce = '{0}\nBcc: {1}@{2}'.format(announce, upload.changes.source, tracking)
 165         subst['__ANNOUNCE_LIST_ADDRESS__'] = announce
 166         message = utils.TemplateSubst(subst, os.path.join(cnf['Dir::Templates'], 'process-unchecked.announce'))
 167         utils.send_mail(message)
 168
 169     # TODO: code duplication. Similar code is in process-upload.
 170     if cnf.find_b('Dinstall::CloseBugs') and upload.changes.closes is not None and upload.source is not None:
 171         for bugnum in upload.changes.closes:
 172             subst['__BUG_NUMBER__'] = bugnum
 173             message = utils.TemplateSubst(subst, os.path.join(cnf['Dir::Templates'], 'process-unchecked.bug-close'))
 174             utils.send_mail(message)
 175
 176             del subst['__BUG_NUMBER__']
 177
 178     # TODO: code duplication. Similar code is in process-upload.
 179     # Move .changes to done
 180     src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 181     now = datetime.datetime.now()
 182     donedir = os.path.join(cnf['Dir::Done'], now.strftime('%Y/%m/%d'))
 183     dst = os.path.join(donedir, upload.changes.changesname)
 184     dst = utils.find_next_free(dst)
 185     fs.copy(src, dst, mode=0o644)
 186
 187     remove_upload(upload, transaction)
 188
 189 ################################################################################
 190
 191 @try_or_reject
 192 def comment_reject(*args):
 193     real_comment_reject(*args)
 194
 195 def real_comment_reject(upload, srcqueue, comments, transaction, notify=True):
 196     cnf = Config()
 197
 198     fs = transaction.fs
 199     session = transaction.session
 200     changesname = upload.changes.changesname
 201     queuedir = upload.policy_queue.path
 202     rejectdir = cnf['Dir::Reject']
 203
 204     ### Copy files to reject/
 205
 206     poolfiles = [b.poolfile for b in upload.binaries]
 207     if upload.source is not None:
 208         poolfiles.extend([df.poolfile for df in upload.source.srcfiles])
 209     # Not beautiful...
 210     files = [ af.path for af in session.query(ArchiveFile) \
 211                   .filter_by(archive=upload.policy_queue.suite.archive) \
 212                   .join(ArchiveFile.file) \
 213                   .filter(PoolFile.file_id.in_([ f.file_id for f in poolfiles ])) ]
 214     for byhand in upload.byhand:
 215         path = os.path.join(queuedir, byhand.filename)
 216         if os.path.exists(path):
 217             files.append(path)
 218     files.append(os.path.join(queuedir, changesname))
 219
 220     for fn in files:
 221         dst = utils.find_next_free(os.path.join(rejectdir, os.path.basename(fn)))
 222         fs.copy(fn, dst, link=True)
 223
 224     ### Write reason
 225
 226     dst = utils.find_next_free(os.path.join(rejectdir, '{0}.reason'.format(changesname)))
 227     fh = fs.create(dst)
 228     fh.write(comments)
 229     fh.close()
 230
 231     ### Send mail notification
 232
 233     if notify:
 234         subst = subst_for_upload(upload)
 235         subst['__MANUAL_REJECT_MESSAGE__'] = ''
 236         subst['__REJECT_MESSAGE__'] = comments
 237
 238         # Try to use From: from comment file if there is one.
 239         # This is not very elegant...
 240         match = re.match(r"\AFrom: ([^\n]+)\n\n", comments)
 241         if match:
 242             subst['__REJECTOR_ADDRESS__'] = match.group(1)
 243             subst['__REJECT_MESSAGE__'] = '\n'.join(comments.splitlines()[2:])
 244
 245         message = utils.TemplateSubst(subst, os.path.join(cnf['Dir::Templates'], 'queue.rejected'))
 246         utils.send_mail(message)
 247
 248     print "  REJECT"
 249     if not Options["No-Action"]:
 250         Logger.log(["Policy Queue REJECT", srcqueue.queue_name, upload.changes.changesname])
 251
 252     remove_upload(upload, transaction)
 253
 254 ################################################################################
 255
 256 def remove_upload(upload, transaction):
 257     fs = transaction.fs
 258     session = transaction.session
 259     changes = upload.changes
 260
 261     # Remove byhand and changes files. Binary and source packages will be
 262     # removed from {bin,src}_associations and eventually removed by clean-suites automatically.
 263     queuedir = upload.policy_queue.path
 264     for byhand in upload.byhand:
 265         path = os.path.join(queuedir, byhand.filename)
 266         if os.path.exists(path):
 267             fs.unlink(path)
 268         session.delete(byhand)
 269     fs.unlink(os.path.join(queuedir, upload.changes.changesname))
 270
 271     session.delete(upload)
 272     session.delete(changes)
 273     session.flush()
 274
 275 ################################################################################
 276
 277 def subst_for_upload(upload):
 278     # TODO: similar code in process-upload
 279     cnf = Config()
 280
 281     maintainer_field = upload.changes.changedby or upload.changes.maintainer
 282     if upload.source is not None:
 283         addresses = utils.mail_addresses_for_upload(upload.changes.maintainer, maintainer_field, upload.changes.fingerprint)
 284     else:
 285         addresses = utils.mail_addresses_for_upload(upload.changes.maintainer, upload.changes.maintainer, upload.changes.fingerprint)
 286
 287     changes_path = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 288     changes_contents = open(changes_path, 'r').read()
 289
 290     bcc = 'X-DAK: dak process-policy'
 291     if 'Dinstall::Bcc' in cnf:
 292         bcc = '{0}\nBcc: {1}'.format(bcc, cnf['Dinstall::Bcc'])
 293
 294     subst = {
 295         '__DISTRO__': cnf['Dinstall::MyDistribution'],
 296         '__ADMIN_ADDRESS__': cnf['Dinstall::MyAdminAddress'],
 297
 298         '__CHANGES_FILENAME__': upload.changes.changesname,
 299         '__SOURCE__': upload.changes.source,
 300         '__VERSION__': upload.changes.version,
 301         '__ARCHITECTURE__': upload.changes.architecture,
 302         '__MAINTAINER__': maintainer_field,
 303         '__MAINTAINER_FROM__': fix_maintainer(maintainer_field)[1],
 304         '__MAINTAINER_TO__': ", ".join(addresses),
 305         '__CC__': 'X-DAK-Rejection: manual or automatic',
 306         '__REJECTOR_ADDRESS__': cnf['Dinstall::MyEmailAddress'],
 307         '__BCC__': bcc,
 308         '__BUG_SERVER__': cnf.get('Dinstall::BugServer'),
 309         '__FILE_CONTENTS__': changes_contents,
 310     }
 311
 312     override_maintainer = cnf.get('Dinstall::OverrideMaintainer')
 313     if override_maintainer:
 314         subst['__MAINTAINER_TO__'] = override_maintainer
 315
 316     return subst
 317
 318 ################################################################################
 319
 320 def remove_unreferenced_binaries(policy_queue, transaction):
 321     """Remove binaries that are no longer referenced by an upload
 322
 323     @type  policy_queue: L{daklib.dbconn.PolicyQueue}
 324
 325     @type  transaction: L{daklib.archive.ArchiveTransaction}
 326     """
 327     session = transaction.session
 328     suite = policy_queue.suite
 329
 330     query = """
 331        SELECT b.*
 332          FROM binaries b
 333          JOIN bin_associations ba ON b.id = ba.bin
 334         WHERE ba.suite = :suite_id
 335           AND NOT EXISTS (SELECT 1 FROM policy_queue_upload_binaries_map pqubm
 336                                    JOIN policy_queue_upload pqu ON pqubm.policy_queue_upload_id = pqu.id
 337                                   WHERE pqu.policy_queue_id = :policy_queue_id
 338                                     AND pqubm.binary_id = b.id)"""
 339     binaries = session.query(DBBinary).from_statement(query) \
 340         .params({'suite_id': policy_queue.suite_id, 'policy_queue_id': policy_queue.policy_queue_id})
 341
 342     for binary in binaries:
 343         Logger.log(["removed binary from policy queue", policy_queue.queue_name, binary.package, binary.version])
 344         transaction.remove_binary(binary, suite)
 345
 346 def remove_unreferenced_sources(policy_queue, transaction):
 347     """Remove sources that are no longer referenced by an upload or a binary
 348
 349     @type  policy_queue: L{daklib.dbconn.PolicyQueue}
 350
 351     @type  transaction: L{daklib.archive.ArchiveTransaction}
 352     """
 353     session = transaction.session
 354     suite = policy_queue.suite
 355
 356     query = """
 357        SELECT s.*
 358          FROM source s
 359          JOIN src_associations sa ON s.id = sa.source
 360         WHERE sa.suite = :suite_id
 361           AND NOT EXISTS (SELECT 1 FROM policy_queue_upload pqu
 362                                   WHERE pqu.policy_queue_id = :policy_queue_id
 363                                     AND pqu.source_id = s.id)
 364           AND NOT EXISTS (SELECT 1 FROM binaries b
 365                                    JOIN bin_associations ba ON b.id = ba.bin
 366                                   WHERE b.source = s.id
 367                                     AND ba.suite = :suite_id)"""
 368     sources = session.query(DBSource).from_statement(query) \
 369         .params({'suite_id': policy_queue.suite_id, 'policy_queue_id': policy_queue.policy_queue_id})
 370
 371     for source in sources:
 372         Logger.log(["removed source from policy queue", policy_queue.queue_name, source.source, source.version])
 373         transaction.remove_source(source, suite)
 374
 375 ################################################################################
 376
 377 def main():
 378     global Options, Logger
 379
 380     cnf = Config()
 381     session = DBConn().session()
 382
 383     Arguments = [('h',"help","Process-Policy::Options::Help"),
 384                  ('n',"no-action","Process-Policy::Options::No-Action")]
 385
 386     for i in ["help", "no-action"]:
 387         if not cnf.has_key("Process-Policy::Options::%s" % (i)):
 388             cnf["Process-Policy::Options::%s" % (i)] = ""
 389
 390     queue_name = apt_pkg.parse_commandline(cnf.Cnf,Arguments,sys.argv)
 391
 392     if len(queue_name) != 1:
 393         print "E: Specify exactly one policy queue"
 394         sys.exit(1)
 395
 396     queue_name = queue_name[0]
 397
 398     Options = cnf.subtree("Process-Policy::Options")
 399
 400     if Options["Help"]:
 401         usage()
 402
 403     Logger = daklog.Logger("process-policy")
 404     if not Options["No-Action"]:
 405         urgencylog = UrgencyLog()
 406
 407     with ArchiveTransaction() as transaction:
 408         session = transaction.session
 409         try:
 410             pq = session.query(PolicyQueue).filter_by(queue_name=queue_name).one()
 411         except NoResultFound:
 412             print "E: Cannot find policy queue %s" % queue_name
 413             sys.exit(1)
 414
 415         commentsdir = os.path.join(pq.path, 'COMMENTS')
 416         # The comments stuff relies on being in the right directory
 417         os.chdir(pq.path)
 418
 419         do_comments(commentsdir, pq, "ACCEPT.", "ACCEPTED.", "OK", comment_accept, transaction)
 420         do_comments(commentsdir, pq, "ACCEPTED.", "ACCEPTED.", "OK", comment_accept, transaction)
 421         do_comments(commentsdir, pq, "REJECT.", "REJECTED.", "NOTOK", comment_reject, transaction)
 422
 423         remove_unreferenced_binaries(pq, transaction)
 424         remove_unreferenced_sources(pq, transaction)
 425
 426     if not Options['No-Action']:
 427         urgencylog.close()
 428
 429 ################################################################################
 430
 431 if __name__ == '__main__':
 432     main()