from daklib.textutils import fix_maintainer, ParseMaintError
import daklib.lintian as lintian
import daklib.utils as utils
+from daklib.upload import InvalidHashException
import apt_inst
import apt_pkg
from apt_pkg import version_compare
+import errno
import os
import time
import yaml
def __str__(self):
return "'%s' has mismatching %s from the external files db ('%s' [current] vs '%s' [external])" % self.args[:4]
+class RejectACL(Reject):
+ """exception raise by failing ACL checks"""
+ def __init__(self, acl, reason):
+ self.acl = acl
+ self.reason = reason
+
+ def __str__(self):
+ return "ACL {0}: {1}".format(self.acl.name, self.reason)
+
class Check(object):
"""base class for checks
class HashesCheck(Check):
"""Check hashes in .changes and .dsc are valid."""
def check(self, upload):
- changes = upload.changes
- for f in changes.files.itervalues():
- f.check(upload.directory)
- source = changes.source
- if source is not None:
- for f in source.files.itervalues():
+ what = None
+ try:
+ changes = upload.changes
+ what = changes.filename
+ for f in changes.files.itervalues():
f.check(upload.directory)
+ source = changes.source
+ if source is not None:
+ what = source.filename
+ for f in source.files.itervalues():
+ f.check(upload.directory)
+ except IOError as e:
+ if e.errno == errno.ENOENT:
+ raise Reject('{0} refers to non-existing file: {1}\n'
+ 'Perhaps you need to include it in your upload?'
+ .format(what, os.path.basename(e.filename)))
+ raise
+ except InvalidHashException as e:
+ raise Reject('{0}: {1}'.format(what, unicode(e)))
class ExternalHashesCheck(Check):
"""Checks hashes in .changes and .dsc against an external database."""
# XXX: Drop DMUA part here and switch to new implementation.
# XXX: Send warning mail once users can set the new DMUA flag
dmua_status, dmua_reason = self._check_dmua(upload)
- if not dmua_status:
- return False, dmua_reason
+ if acl_per_source is None:
+ if not dmua_status:
+ return False, dmua_reason
+ else:
+ upload.warn('DM flag not set, but accepted as DMUA was set.')
#if acl_per_source is None:
# return False, "not allowed to upload source package '{0}'".format(source_name)
if acl.deny_per_source and acl_per_source is not None:
raise Reject('No ACL for fingerprint {0}'.format(fingerprint.fingerprint))
result, reason = self._check_acl(session, upload, acl)
if not result:
- raise Reject(reason)
+ raise RejectACL(acl, reason)
for acl in session.query(ACL).filter_by(is_global=True):
result, reason = self._check_acl(session, upload, acl)
if result == False:
- raise Reject(reason)
+ raise RejectACL(acl, reason)
return True