"""module provided pre-acceptance tests
-Please read the documentation for the `Check` class for the interface.
+Please read the documentation for the L{Check} class for the interface.
"""
from daklib.config import Config
-from .dbconn import *
+from daklib.dbconn import *
import daklib.dbconn as dbconn
-from .regexes import *
-from .textutils import fix_maintainer, ParseMaintError
+from daklib.regexes import *
+from daklib.textutils import fix_maintainer, ParseMaintError
import daklib.lintian as lintian
import daklib.utils as utils
class Check(object):
"""base class for checks
- checks are called by daklib.archive.ArchiveUpload. Failing tests should
- raise a `daklib.checks.Reject` exception including a human-readable
+ checks are called by L{daklib.archive.ArchiveUpload}. Failing tests should
+ raise a L{daklib.checks.Reject} exception including a human-readable
description why the upload should be rejected.
"""
def check(self, upload):
"""do checks
- Args:
- upload (daklib.archive.ArchiveUpload): upload to check
+ @type upload: L{daklib.archive.ArchiveUpload}
+ @param upload: upload to check
- Raises:
- daklib.checks.Reject
+ @raise daklib.checks.Reject: upload should be rejected
"""
raise NotImplemented
def per_suite_check(self, upload, suite):
"""do per-suite checks
- Args:
- upload (daklib.archive.ArchiveUpload): upload to check
- suite (daklib.dbconn.Suite): suite to check
+ @type upload: L{daklib.archive.ArchiveUpload}
+ @param upload: upload to check
- Raises:
- daklib.checks.Reject
+ @type suite: L{daklib.dbconn.Suite}
+ @param suite: suite to check
+
+ @raise daklib.checks.Reject: upload should be rejected
"""
raise NotImplemented
@property
def forcable(self):
"""allow to force ignore failing test
- True if it is acceptable to force ignoring a failing test,
- False otherwise
+ C{True} if it is acceptable to force ignoring a failing test,
+ C{False} otherwise
"""
return False
except Exception as e:
raise Reject('{0}: APT could not parse {1} field: {2}'.format(dsc_fn, field, e))
- # TODO: check all expected files for given source format are included
+ rejects = utils.check_dsc_files(dsc_fn, control, source.files.keys())
+ if len(rejects) > 0:
+ raise Reject("\n".join(rejects))
+
+ return True
class SingleDistributionCheck(Check):
"""Check that the .changes targets only a single distribution."""
class ACLCheck(Check):
"""Check the uploader is allowed to upload the packages in .changes"""
- def _check_dm(self, upload):
+
+ def _does_hijack(self, session, upload, suite):
+ # Try to catch hijacks.
+ # This doesn't work correctly. Uploads to experimental can still
+ # "hijack" binaries from unstable. Also one can hijack packages
+ # via buildds (but people who try this should not be DMs).
+ for binary_name in upload.changes.binary_names:
+ binaries = session.query(DBBinary).join(DBBinary.source) \
+ .filter(DBBinary.suites.contains(suite)) \
+ .filter(DBBinary.package == binary_name)
+ for binary in binaries:
+ if binary.source.source != upload.changes.changes['Source']:
+ return True, binary, binary.source.source
+ return False, None, None
+
+ def _check_acl(self, session, upload, acl):
+ source_name = upload.changes.source_name
+
+ if acl.match_fingerprint and upload.fingerprint not in acl.fingerprints:
+ return None, None
+ if acl.match_keyring is not None and upload.fingerprint.keyring != acl.match_keyring:
+ return None, None
+
+ if not acl.allow_new:
+ if upload.new:
+ return False, "NEW uploads are not allowed"
+ for f in upload.changes.files.itervalues():
+ if f.section == 'byhand' or f.section.startswith("raw-"):
+ return False, "BYHAND uploads are not allowed"
+ if not acl.allow_source and upload.changes.source is not None:
+ return False, "sourceful uploads are not allowed"
+ binaries = upload.changes.binaries
+ if len(binaries) != 0:
+ if not acl.allow_binary:
+ return False, "binary uploads are not allowed"
+ if upload.changes.source is None and not acl.allow_binary_only:
+ return False, "binary-only uploads are not allowed"
+ if not acl.allow_binary_all:
+ uploaded_arches = set(upload.changes.architectures)
+ uploaded_arches.discard('source')
+ allowed_arches = set(a.arch_string for a in acl.architectures)
+ forbidden_arches = uploaded_arches - allowed_arches
+ if len(forbidden_arches) != 0:
+ return False, "uploads for architecture(s) {0} are not allowed".format(", ".join(forbidden_arches))
+ if not acl.allow_hijack:
+ for suite in upload.final_suites:
+ does_hijack, hijacked_binary, hijacked_from = self._does_hijack(session, upload, suite)
+ if does_hijack:
+ return False, "hijacks are not allowed (binary={0}, other-source={1})".format(hijacked_binary, hijacked_from)
+
+ acl_per_source = session.query(ACLPerSource).filter_by(acl=acl, fingerprint=upload.fingerprint, source=source_name).first()
+ if acl.allow_per_source:
+ # 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:
+ # return False, "not allowed to upload source package '{0}'".format(source_name)
+ if acl.deny_per_source and acl_per_source is not None:
+ return False, acl_per_source.reason or "forbidden to upload source package '{0}'".format(source_name)
+
+ return True, None
+
+ def _check_dmua(self, upload):
# This code is not very nice, but hopefully works until we can replace
# DM-Upload-Allowed, cf. https://lists.debian.org/debian-project/2012/06/msg00029.html
session = upload.session
- if 'source' not in upload.changes.architectures:
- raise Reject('DM uploads must include source')
- for f in upload.changes.files.itervalues():
- if f.section == 'byhand' or f.section[:4] == "raw-":
- raise Reject("Uploading byhand packages is not allowed for DMs.")
-
- # Reject NEW packages
- distributions = upload.changes.distributions
- assert len(distributions) == 1
- suite = session.query(Suite).filter_by(suite_name=distributions[0]).one()
- overridesuite = suite
- if suite.overridesuite is not None:
- overridesuite = session.query(Suite).filter_by(suite_name=suite.overridesuite).one()
- if upload._check_new(overridesuite):
- raise Reject('Uploading NEW packages is not allowed for DMs.')
-
# Check DM-Upload-Allowed
+ suites = upload.final_suites
+ assert len(suites) == 1
+ suite = list(suites)[0]
+
last_suites = ['unstable', 'experimental']
if suite.suite_name.endswith('-backports'):
last_suites = [suite.suite_name]
.join(DBSource.suites).filter(Suite.suite_name.in_(last_suites)) \
.order_by(DBSource.version.desc()).limit(1).first()
if last is None:
- raise Reject('No existing source found in {0}'.format(' or '.join(last_suites)))
+ return False, 'No existing source found in {0}'.format(' or '.join(last_suites))
if not last.dm_upload_allowed:
- raise Reject('DM-Upload-Allowed is not set in {0}={1}'.format(last.source, last.version))
+ return False, 'DM-Upload-Allowed is not set in {0}={1}'.format(last.source, last.version)
# check current Changed-by is in last Maintainer or Uploaders
uploader_names = [ u.name for u in last.uploaders ]
changed_by_field = upload.changes.changes.get('Changed-By', upload.changes.changes['Maintainer'])
if changed_by_field not in uploader_names:
- raise Reject('{0} is not an uploader for {1}={2}'.format(changed_by_field, last.source, last.version))
+ return False, '{0} is not an uploader for {1}={2}'.format(changed_by_field, last.source, last.version)
# check Changed-by is the DM
changed_by = fix_maintainer(changed_by_field)
uid = upload.fingerprint.uid
if uid is None:
- raise Reject('Unknown uid for fingerprint {0}'.format(upload.fingerprint.fingerprint))
+ return False, 'Unknown uid for fingerprint {0}'.format(upload.fingerprint.fingerprint)
if uid.uid != changed_by[3] and uid.name != changed_by[2]:
- raise Reject('DMs are not allowed to sponsor uploads (expected {0} <{1}> as maintainer, but got {2})'.format(uid.name, uid.uid, changed_by_field))
+ return False, 'DMs are not allowed to sponsor uploads (expected {0} <{1}> as maintainer, but got {2})'.format(uid.name, uid.uid, changed_by_field)
- # Try to catch hijacks.
- # This doesn't work correctly. Uploads to experimental can still
- # "hijack" binaries from unstable. Also one can hijack packages
- # via buildds (but people who try this should not be DMs).
- for binary_name in upload.changes.binary_names:
- binaries = session.query(DBBinary).join(DBBinary.source) \
- .join(DBBinary.suites).filter(Suite.suite_name.in_(upload.changes.distributions)) \
- .filter(DBBinary.package == binary_name)
- for binary in binaries:
- if binary.source.source != upload.changes.changes['Source']:
- raise Reject('DMs must not hijack binaries (binary={0}, other-source={1})'.format(binary_name, binary.source.source))
+ return True, None
+
+ def check(self, upload):
+ session = upload.session
+ fingerprint = upload.fingerprint
+ keyring = fingerprint.keyring
+
+ if keyring is None:
+ raise Reject('No keyring for fingerprint {0}'.format(fingerprint.fingerprint))
+ if not keyring.active:
+ raise Reject('Keyring {0} is not active'.format(keyring.name))
+
+ acl = fingerprint.acl or keyring.acl
+ if acl is 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)
+
+ 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)
return True
+ def per_suite_check(self, upload, suite):
+ acls = suite.acls
+ if len(acls) != 0:
+ accept = False
+ for acl in acls:
+ result, reason = self._check_acl(upload.session, upload, acl)
+ if result == False:
+ raise Reject(reason)
+ accept = accept or result
+ if not accept:
+ raise Reject('Not accepted by any per-suite acl (suite={0})'.format(suite.suite_name))
+ return True
+
+class TransitionCheck(Check):
+ """check for a transition"""
def check(self, upload):
- fingerprint = upload.fingerprint
- source_acl = fingerprint.source_acl
- if source_acl is None:
- if 'source' in upload.changes.architectures:
- raise Reject('Fingerprint {0} must not upload source'.format(fingerprint.fingerprint))
- elif source_acl.access_level == 'dm':
- self._check_dm(upload)
- elif source_acl.access_level != 'full':
- raise Reject('Unknown source_acl access level {0} for fingerprint {1}'.format(source_acl.access_level, fingerprint.fingerprint))
-
- bin_architectures = set(upload.changes.architectures)
- bin_architectures.discard('source')
- binary_acl = fingerprint.binary_acl
- if binary_acl is None:
- if len(bin_architectures) > 0:
- raise Reject('Fingerprint {0} must not upload binary packages'.format(fingerprint.fingerprint))
- elif binary_acl.access_level == 'map':
- query = upload.session.query(BinaryACLMap).filter_by(fingerprint=fingerprint)
- allowed_architectures = [ m.architecture.arch_string for m in query ]
-
- for arch in upload.changes.architectures:
- if arch not in allowed_architectures:
- raise Reject('Fingerprint {0} must not upload binaries for architecture {1}'.format(fingerprint.fingerprint, arch))
- elif binary_acl.access_level != 'full':
- raise Reject('Unknown binary_acl access level {0} for fingerprint {1}'.format(binary_acl.access_level, fingerprint.fingerprint))
+ if 'source' not in upload.changes.architectures:
+ return True
+
+ transitions = self.get_transitions()
+ if transitions is None:
+ return True
+
+ control = upload.changes.changes
+ source = re_field_source.match(control['Source']).group('package')
+
+ for trans in transitions:
+ t = transitions[trans]
+ source = t["source"]
+ expected = t["new"]
+
+ # Will be None if nothing is in testing.
+ current = get_source_in_suite(source, "testing", session)
+ if current is not None:
+ compare = apt_pkg.version_compare(current.version, expected)
+
+ if current is None or compare < 0:
+ # This is still valid, the current version in testing is older than
+ # the new version we wait for, or there is none in testing yet
+
+ # Check if the source we look at is affected by this.
+ if source in t['packages']:
+ # The source is affected, lets reject it.
+
+ rejectmsg = "{0}: part of the {1} transition.\n\n".format(source, trans)
+
+ if current is not None:
+ currentlymsg = "at version {0}".format(current.version)
+ else:
+ currentlymsg = "not present in testing"
+
+ rejectmsg += "Transition description: {0}\n\n".format(t["reason"])
+
+ rejectmsg += "\n".join(textwrap.wrap("""Your package
+is part of a testing transition designed to get {0} migrated (it is
+currently {1}, we need version {2}). This transition is managed by the
+Release Team, and {3} is the Release-Team member responsible for it.
+Please mail debian-release@lists.debian.org or contact {3} directly if you
+need further assistance. You might want to upload to experimental until this
+transition is done.""".format(source, currentlymsg, expected,t["rm"])))
+
+ raise Reject(rejectmsg)
return True
+ def get_transitions(self):
+ cnf = Config()
+ path = cnf.get('Dinstall::ReleaseTransitions', '')
+ if path == '' or not os.path.exists(path):
+ return None
+
+ contents = file(path, 'r').read()
+ try:
+ transitions = yaml.load(contents)
+ return transitions
+ except yaml.YAMLError as msg:
+ utils.warn('Not checking transitions, the transitions file is broken: {0}'.format(msg))
+
+ return None
+
class NoSourceOnlyCheck(Check):
"""Check for source-only upload
except yaml.YAMLError as msg:
raise Exception('Could not read lintian tags file {0}, YAML error: {1}'.format(tagfile, msg))
- fd, temp_filename = utils.temp_filename()
+ fd, temp_filename = utils.temp_filename(mode=0o644)
temptagfile = os.fdopen(fd, 'w')
for tags in lintiantags.itervalues():
for tag in tags:
changespath = os.path.join(upload.directory, changes.filename)
try:
- # FIXME: no shell
- cmd = "lintian --show-overrides --tags-from-file {0} {1}".format(temp_filename, changespath)
+ if cnf.unpribgroup:
+ cmd = "sudo -H -u {0} -- /usr/bin/lintian --show-overrides --tags-from-file {1} {2}".format(cnf.unprivgroup, temp_filename, changespath)
+ else:
+ cmd = "/usr/bin/lintian --show-overrides --tags-from-file {0} {1}".format(temp_filename, changespath)
result, output = commands.getstatusoutput(cmd)
finally:
os.unlink(temp_filename)
projects / dak.git / blobdiff