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 from sqlalchemy.orm.exc import NoResultFound
  41
  42 from daklib.dbconn import *
  43 from daklib import daklog
  44 from daklib import utils
  45 from daklib.dak_exceptions import CantOpenError, AlreadyLockedError, CantGetLockError
  46 from daklib.config import Config
  47 from daklib.archive import ArchiveTransaction, source_component_from_package_list
  48 from daklib.urgencylog import UrgencyLog
  49 from daklib.packagelist import PackageList
  50
  51 import daklib.announce
  52 import daklib.utils
  53
  54 # Globals
  55 Options = None
  56 Logger = None
  57
  58 ################################################################################
  59
  60 def do_comments(dir, srcqueue, opref, npref, line, fn, transaction):
  61     session = transaction.session
  62     actions = []
  63     for comm in [ x for x in os.listdir(dir) if x.startswith(opref) ]:
  64         lines = open(os.path.join(dir, comm)).readlines()
  65         if len(lines) == 0 or lines[0] != line + "\n": continue
  66
  67         # If the ACCEPT includes a _<arch> we only accept that .changes.
  68         # Otherwise we accept all .changes that start with the given prefix
  69         changes_prefix = comm[len(opref):]
  70         if changes_prefix.count('_') < 2:
  71             changes_prefix = changes_prefix + '_'
  72         else:
  73             changes_prefix = changes_prefix + '.changes'
  74
  75         # We need to escape "_" as we use it with the LIKE operator (via the
  76         # SQLA startwith) later.
  77         changes_prefix = changes_prefix.replace("_", r"\_")
  78
  79         uploads = session.query(PolicyQueueUpload).filter_by(policy_queue=srcqueue) \
  80             .join(PolicyQueueUpload.changes).filter(DBChange.changesname.startswith(changes_prefix)) \
  81             .order_by(PolicyQueueUpload.source_id)
  82         reason = "".join(lines[1:])
  83         actions.extend((u, reason) for u in uploads)
  84
  85         if opref != npref:
  86             newcomm = npref + comm[len(opref):]
  87             newcomm = utils.find_next_free(os.path.join(dir, newcomm))
  88             transaction.fs.move(os.path.join(dir, comm), newcomm)
  89
  90     actions.sort()
  91
  92     for u, reason in actions:
  93         print("Processing changes file: {0}".format(u.changes.changesname))
  94         fn(u, srcqueue, reason, transaction)
  95
  96 ################################################################################
  97
  98 def try_or_reject(function):
  99     def wrapper(upload, srcqueue, comments, transaction):
 100         try:
 101             function(upload, srcqueue, comments, transaction)
 102         except Exception as e:
 103             comments = 'An exception was raised while processing the package:\n{0}\nOriginal comments:\n{1}'.format(traceback.format_exc(), comments)
 104             try:
 105                 transaction.rollback()
 106                 real_comment_reject(upload, srcqueue, comments, transaction)
 107             except Exception as e:
 108                 comments = 'In addition an exception was raised while trying to reject the upload:\n{0}\nOriginal rejection:\n{1}'.format(traceback.format_exc(), comments)
 109                 transaction.rollback()
 110                 real_comment_reject(upload, srcqueue, comments, transaction, notify=False)
 111         if not Options['No-Action']:
 112             transaction.commit()
 113         else:
 114             transaction.rollback()
 115     return wrapper
 116
 117 ################################################################################
 118
 119 @try_or_reject
 120 def comment_accept(upload, srcqueue, comments, transaction):
 121     for byhand in upload.byhand:
 122         path = os.path.join(srcqueue.path, byhand.filename)
 123         if os.path.exists(path):
 124             raise Exception('E: cannot ACCEPT upload with unprocessed byhand file {0}'.format(byhand.filename))
 125
 126     cnf = Config()
 127
 128     fs = transaction.fs
 129     session = transaction.session
 130     changesname = upload.changes.changesname
 131     allow_tainted = srcqueue.suite.archive.tainted
 132
 133     # We need overrides to get the target component
 134     overridesuite = upload.target_suite
 135     if overridesuite.overridesuite is not None:
 136         overridesuite = session.query(Suite).filter_by(suite_name=overridesuite.overridesuite).one()
 137
 138     def binary_component_func(db_binary):
 139         section = db_binary.proxy['Section']
 140         component_name = 'main'
 141         if section.find('/') != -1:
 142             component_name = section.split('/', 1)[0]
 143         return get_mapped_component(component_name, session=session)
 144
 145     def is_debug_binary(db_binary):
 146         return daklib.utils.is_in_debug_section(db_binary.proxy)
 147
 148     def has_debug_binaries(upload):
 149         return any((is_debug_binary(x) for x in upload.binaries))
 150
 151     def source_component_func(db_source):
 152         package_list = PackageList(db_source.proxy)
 153         component = source_component_from_package_list(package_list, upload.target_suite)
 154         if component is not None:
 155             return get_mapped_component(component.component_name, session=session)
 156
 157         # Fallback for packages without Package-List field
 158         query = session.query(Override).filter_by(suite=overridesuite, package=db_source.source) \
 159             .join(OverrideType).filter(OverrideType.overridetype == 'dsc') \
 160             .join(Component)
 161         return query.one().component
 162
 163     all_target_suites = [upload.target_suite]
 164     all_target_suites.extend([q.suite for q in upload.target_suite.copy_queues])
 165
 166     for suite in all_target_suites:
 167         debug_suite = suite.debug_suite
 168
 169         if upload.source is not None:
 170             # If we have Source in this upload, let's include it into
 171             # upload suite.
 172             transaction.copy_source(
 173                 upload.source,
 174                 suite,
 175                 source_component_func(upload.source),
 176                 allow_tainted=allow_tainted,
 177             )
 178
 179             if debug_suite is not None and has_debug_binaries(upload):
 180                 # If we're handing a debug package, we also need to include the
 181                 # source in the debug suite as well.
 182                 transaction.copy_source(
 183                     upload.source,
 184                     debug_suite,
 185                     source_component_func(upload.source),
 186                     allow_tainted=allow_tainted,
 187                 )
 188
 189         for db_binary in upload.binaries:
 190             # Now, let's work out where to copy this guy to -- if it's
 191             # a debug binary, and the suite has a debug suite, let's go
 192             # ahead and target the debug suite rather then the stock
 193             # suite.
 194             copy_to_suite = suite
 195             if debug_suite is not None and is_debug_binary(db_binary):
 196                 copy_to_suite = debug_suite
 197
 198             # build queues and debug suites may miss the source package
 199             # if this is a binary-only upload.
 200             if copy_to_suite != upload.target_suite:
 201                 transaction.copy_source(
 202                     db_binary.source,
 203                     copy_to_suite,
 204                     source_component_func(db_binary.source),
 205                     allow_tainted=allow_tainted,
 206                 )
 207
 208             transaction.copy_binary(
 209                 db_binary,
 210                 copy_to_suite,
 211                 binary_component_func(db_binary),
 212                 allow_tainted=allow_tainted,
 213                 extra_archives=[upload.target_suite.archive],
 214             )
 215
 216         suite.update_last_changed()
 217
 218     # Copy .changes if needed
 219     if upload.target_suite.copychanges:
 220         src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 221         dst = os.path.join(upload.target_suite.path, upload.changes.changesname)
 222         fs.copy(src, dst, mode=upload.target_suite.archive.mode)
 223
 224     # Copy upload to Process-Policy::CopyDir
 225     # Used on security.d.o to sync accepted packages to ftp-master, but this
 226     # should eventually be replaced by something else.
 227     copydir = cnf.get('Process-Policy::CopyDir') or None
 228     if copydir is not None:
 229         mode = upload.target_suite.archive.mode
 230         if upload.source is not None:
 231             for f in [ df.poolfile for df in upload.source.srcfiles ]:
 232                 dst = os.path.join(copydir, f.basename)
 233                 if not os.path.exists(dst):
 234                     fs.copy(f.fullpath, dst, mode=mode)
 235
 236         for db_binary in upload.binaries:
 237             f = db_binary.poolfile
 238             dst = os.path.join(copydir, f.basename)
 239             if not os.path.exists(dst):
 240                 fs.copy(f.fullpath, dst, mode=mode)
 241
 242         src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 243         dst = os.path.join(copydir, upload.changes.changesname)
 244         if not os.path.exists(dst):
 245             fs.copy(src, dst, mode=mode)
 246
 247     if upload.source is not None and not Options['No-Action']:
 248         urgency = upload.changes.urgency
 249         if urgency not in cnf.value_list('Urgency::Valid'):
 250             urgency = cnf['Urgency::Default']
 251         UrgencyLog().log(upload.source.source, upload.source.version, urgency)
 252
 253     print "  ACCEPT"
 254     if not Options['No-Action']:
 255         Logger.log(["Policy Queue ACCEPT", srcqueue.queue_name, changesname])
 256
 257     pu = get_processed_upload(upload)
 258     daklib.announce.announce_accept(pu)
 259
 260     # TODO: code duplication. Similar code is in process-upload.
 261     # Move .changes to done
 262     src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 263     now = datetime.datetime.now()
 264     donedir = os.path.join(cnf['Dir::Done'], now.strftime('%Y/%m/%d'))
 265     dst = os.path.join(donedir, upload.changes.changesname)
 266     dst = utils.find_next_free(dst)
 267     fs.copy(src, dst, mode=0o644)
 268
 269     remove_upload(upload, transaction)
 270
 271 ################################################################################
 272
 273 @try_or_reject
 274 def comment_reject(*args):
 275     real_comment_reject(*args, manual=True)
 276
 277 def real_comment_reject(upload, srcqueue, comments, transaction, notify=True, manual=False):
 278     cnf = Config()
 279
 280     fs = transaction.fs
 281     session = transaction.session
 282     changesname = upload.changes.changesname
 283     queuedir = upload.policy_queue.path
 284     rejectdir = cnf['Dir::Reject']
 285
 286     ### Copy files to reject/
 287
 288     poolfiles = [b.poolfile for b in upload.binaries]
 289     if upload.source is not None:
 290         poolfiles.extend([df.poolfile for df in upload.source.srcfiles])
 291     # Not beautiful...
 292     files = [ af.path for af in session.query(ArchiveFile) \
 293                   .filter_by(archive=upload.policy_queue.suite.archive) \
 294                   .join(ArchiveFile.file) \
 295                   .filter(PoolFile.file_id.in_([ f.file_id for f in poolfiles ])) ]
 296     for byhand in upload.byhand:
 297         path = os.path.join(queuedir, byhand.filename)
 298         if os.path.exists(path):
 299             files.append(path)
 300     files.append(os.path.join(queuedir, changesname))
 301
 302     for fn in files:
 303         dst = utils.find_next_free(os.path.join(rejectdir, os.path.basename(fn)))
 304         fs.copy(fn, dst, link=True)
 305
 306     ### Write reason
 307
 308     dst = utils.find_next_free(os.path.join(rejectdir, '{0}.reason'.format(changesname)))
 309     fh = fs.create(dst)
 310     fh.write(comments)
 311     fh.close()
 312
 313     ### Send mail notification
 314
 315     if notify:
 316         rejected_by = None
 317         reason = comments
 318
 319         # Try to use From: from comment file if there is one.
 320         # This is not very elegant...
 321         match = re.match(r"\AFrom: ([^\n]+)\n\n", comments)
 322         if match:
 323             rejected_by = match.group(1)
 324             reason = '\n'.join(comments.splitlines()[2:])
 325
 326         pu = get_processed_upload(upload)
 327         daklib.announce.announce_reject(pu, reason, rejected_by)
 328
 329     print "  REJECT"
 330     if not Options["No-Action"]:
 331         Logger.log(["Policy Queue REJECT", srcqueue.queue_name, upload.changes.changesname])
 332
 333     changes = upload.changes
 334     remove_upload(upload, transaction)
 335     session.delete(changes)
 336
 337 ################################################################################
 338
 339 def remove_upload(upload, transaction):
 340     fs = transaction.fs
 341     session = transaction.session
 342     changes = upload.changes
 343
 344     # Remove byhand and changes files. Binary and source packages will be
 345     # removed from {bin,src}_associations and eventually removed by clean-suites automatically.
 346     queuedir = upload.policy_queue.path
 347     for byhand in upload.byhand:
 348         path = os.path.join(queuedir, byhand.filename)
 349         if os.path.exists(path):
 350             fs.unlink(path)
 351         session.delete(byhand)
 352     fs.unlink(os.path.join(queuedir, upload.changes.changesname))
 353
 354     session.delete(upload)
 355     session.flush()
 356
 357 ################################################################################
 358
 359 def get_processed_upload(upload):
 360     pu = daklib.announce.ProcessedUpload()
 361
 362     pu.maintainer = upload.changes.maintainer
 363     pu.changed_by = upload.changes.changedby
 364     pu.fingerprint = upload.changes.fingerprint
 365
 366     pu.suites = [ upload.target_suite ]
 367     pu.from_policy_suites = [ upload.target_suite ]
 368
 369     changes_path = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 370     pu.changes = open(changes_path, 'r').read()
 371     pu.changes_filename = upload.changes.changesname
 372     pu.sourceful = upload.source is not None
 373     pu.source = upload.changes.source
 374     pu.version = upload.changes.version
 375     pu.architecture = upload.changes.architecture
 376     pu.bugs = upload.changes.closes
 377
 378     pu.program = "process-policy"
 379
 380     return pu
 381
 382 ################################################################################
 383
 384 def remove_unreferenced_binaries(policy_queue, transaction):
 385     """Remove binaries that are no longer referenced by an upload
 386
 387     @type  policy_queue: L{daklib.dbconn.PolicyQueue}
 388
 389     @type  transaction: L{daklib.archive.ArchiveTransaction}
 390     """
 391     session = transaction.session
 392     suite = policy_queue.suite
 393
 394     query = """
 395        SELECT b.*
 396          FROM binaries b
 397          JOIN bin_associations ba ON b.id = ba.bin
 398         WHERE ba.suite = :suite_id
 399           AND NOT EXISTS (SELECT 1 FROM policy_queue_upload_binaries_map pqubm
 400                                    JOIN policy_queue_upload pqu ON pqubm.policy_queue_upload_id = pqu.id
 401                                   WHERE pqu.policy_queue_id = :policy_queue_id
 402                                     AND pqubm.binary_id = b.id)"""
 403     binaries = session.query(DBBinary).from_statement(query) \
 404         .params({'suite_id': policy_queue.suite_id, 'policy_queue_id': policy_queue.policy_queue_id})
 405
 406     for binary in binaries:
 407         Logger.log(["removed binary from policy queue", policy_queue.queue_name, binary.package, binary.version])
 408         transaction.remove_binary(binary, suite)
 409
 410 def remove_unreferenced_sources(policy_queue, transaction):
 411     """Remove sources that are no longer referenced by an upload or a binary
 412
 413     @type  policy_queue: L{daklib.dbconn.PolicyQueue}
 414
 415     @type  transaction: L{daklib.archive.ArchiveTransaction}
 416     """
 417     session = transaction.session
 418     suite = policy_queue.suite
 419
 420     query = """
 421        SELECT s.*
 422          FROM source s
 423          JOIN src_associations sa ON s.id = sa.source
 424         WHERE sa.suite = :suite_id
 425           AND NOT EXISTS (SELECT 1 FROM policy_queue_upload pqu
 426                                   WHERE pqu.policy_queue_id = :policy_queue_id
 427                                     AND pqu.source_id = s.id)
 428           AND NOT EXISTS (SELECT 1 FROM binaries b
 429                                    JOIN bin_associations ba ON b.id = ba.bin
 430                                   WHERE b.source = s.id
 431                                     AND ba.suite = :suite_id)"""
 432     sources = session.query(DBSource).from_statement(query) \
 433         .params({'suite_id': policy_queue.suite_id, 'policy_queue_id': policy_queue.policy_queue_id})
 434
 435     for source in sources:
 436         Logger.log(["removed source from policy queue", policy_queue.queue_name, source.source, source.version])
 437         transaction.remove_source(source, suite)
 438
 439 ################################################################################
 440
 441 def main():
 442     global Options, Logger
 443
 444     cnf = Config()
 445     session = DBConn().session()
 446
 447     Arguments = [('h',"help","Process-Policy::Options::Help"),
 448                  ('n',"no-action","Process-Policy::Options::No-Action")]
 449
 450     for i in ["help", "no-action"]:
 451         if not cnf.has_key("Process-Policy::Options::%s" % (i)):
 452             cnf["Process-Policy::Options::%s" % (i)] = ""
 453
 454     queue_name = apt_pkg.parse_commandline(cnf.Cnf,Arguments,sys.argv)
 455
 456     if len(queue_name) != 1:
 457         print "E: Specify exactly one policy queue"
 458         sys.exit(1)
 459
 460     queue_name = queue_name[0]
 461
 462     Options = cnf.subtree("Process-Policy::Options")
 463
 464     if Options["Help"]:
 465         usage()
 466
 467     Logger = daklog.Logger("process-policy")
 468     if not Options["No-Action"]:
 469         urgencylog = UrgencyLog()
 470
 471     with ArchiveTransaction() as transaction:
 472         session = transaction.session
 473         try:
 474             pq = session.query(PolicyQueue).filter_by(queue_name=queue_name).one()
 475         except NoResultFound:
 476             print "E: Cannot find policy queue %s" % queue_name
 477             sys.exit(1)
 478
 479         commentsdir = os.path.join(pq.path, 'COMMENTS')
 480         # The comments stuff relies on being in the right directory
 481         os.chdir(pq.path)
 482
 483         do_comments(commentsdir, pq, "REJECT.", "REJECTED.", "NOTOK", comment_reject, transaction)
 484         do_comments(commentsdir, pq, "ACCEPT.", "ACCEPTED.", "OK", comment_accept, transaction)
 485         do_comments(commentsdir, pq, "ACCEPTED.", "ACCEPTED.", "OK", comment_accept, transaction)
 486
 487         remove_unreferenced_binaries(pq, transaction)
 488         remove_unreferenced_sources(pq, transaction)
 489
 490     if not Options['No-Action']:
 491         urgencylog.close()
 492
 493 ################################################################################
 494
 495 if __name__ == '__main__':
 496     main()