Coverage for daklib/checks.py: 70%
664 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1# Copyright (C) 2012, Ansgar Burchardt <ansgar@debian.org>
2#
3# Parts based on code that is
4# Copyright (C) 2001-2006, James Troup <james@nocrew.org>
5# Copyright (C) 2009-2010, Joerg Jaspert <joerg@debian.org>
6#
7# This program is free software; you can redistribute it and/or modify
8# it under the terms of the GNU General Public License as published by
9# the Free Software Foundation; either version 2 of the License, or
10# (at your option) any later version.
11#
12# This program is distributed in the hope that it will be useful,
13# but WITHOUT ANY WARRANTY; without even the implied warranty of
14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15# GNU General Public License for more details.
16#
17# You should have received a copy of the GNU General Public License along
18# with this program; if not, write to the Free Software Foundation, Inc.,
19# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
21"""module provided pre-acceptance tests
23Please read the documentation for the :class:`Check` class for the interface.
24"""
26import datetime
27import os
28import re
29import subprocess
30import tempfile
31import textwrap
32import time
33from collections.abc import Callable, Iterable
34from typing import TYPE_CHECKING, Literal, cast, override
36import apt_inst
37import apt_pkg
38import yaml
39from apt_pkg import version_compare
40from sqlalchemy import select, sql
42import daklib.gpg
43import daklib.upload
44from daklib import dbconn, lintian, utils
45from daklib.config import Config
46from daklib.dbconn import (
47 ACL,
48 ACLPerSource,
49 Architecture,
50 DBBinary,
51 DBSource,
52 SignatureHistory,
53 SrcFormat,
54 Suite,
55 get_source_in_suite,
56)
57from daklib.regexes import (
58 re_field_package,
59 re_field_source,
60 re_field_version,
61 re_field_version_upstream,
62 re_file_binary,
63 re_file_changes,
64 re_file_dsc,
65 re_file_orig,
66 re_file_source,
67 re_isanum,
68)
69from daklib.textutils import ParseMaintError, fix_maintainer
71if TYPE_CHECKING:
72 from sqlalchemy.orm import Session
74 import daklib.archive
77def check_fields_for_valid_utf8(filename: str, control: apt_pkg.TagSection) -> None:
78 """Check all fields of a control file for valid UTF-8"""
79 for field in control.keys():
80 try:
81 # Access the field value to make `TagSection` try to decode it.
82 # We should also do the same for the field name, but this requires
83 # https://bugs.debian.org/995118 to be fixed.
84 # TODO: make sure the field name `field` is valid UTF-8 too
85 control[field]
86 except UnicodeDecodeError:
87 raise Reject(
88 "{0}: The {1} field is not valid UTF-8".format(filename, field)
89 )
92class Reject(Exception):
93 """exception raised by failing checks"""
96class RejectExternalFilesMismatch(Reject):
97 """exception raised by failing the external hashes check"""
99 @override
100 def __str__(self):
101 return (
102 "'%s' has mismatching %s from the external files db ('%s' [current] vs '%s' [external])"
103 % self.args[:4]
104 )
107class RejectACL(Reject):
108 """exception raise by failing ACL checks"""
110 def __init__(self, acl: ACL, reason: str):
111 self.acl = acl
112 self.reason = reason
114 @override
115 def __str__(self):
116 return "ACL {0}: {1}".format(self.acl.name, self.reason)
119class Check:
120 """base class for checks
122 checks are called by :class:`daklib.archive.ArchiveUpload`. Failing tests should
123 raise a :exc:`daklib.checks.Reject` exception including a human-readable
124 description why the upload should be rejected.
125 """
127 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
128 """do checks
130 :param upload: upload to check
132 :raises Reject: upload should be rejected
133 """
134 raise NotImplementedError
136 def per_suite_check(
137 self, upload: "daklib.archive.ArchiveUpload", suite: Suite
138 ) -> bool:
139 """do per-suite checks
141 :param upload: upload to check
142 :param suite: suite to check
144 :raises Reject: upload should be rejected
145 """
146 raise NotImplementedError
148 @property
149 def forcable(self) -> bool:
150 """allow to force ignore failing test
152 :const:`True` if it is acceptable to force ignoring a failing test,
153 :const:`False` otherwise
154 """
155 return False
158class SignatureAndHashesCheck(Check):
159 """Check signature of changes and dsc file (if included in upload)
161 Make sure the signature is valid and done by a known user.
162 """
164 def check_replay(self, upload: "daklib.archive.ArchiveUpload") -> bool:
165 # Use private session as we want to remember having seen the .changes
166 # in all cases.
167 session = upload.session
168 history = SignatureHistory.from_signed_file(upload.changes.signature)
169 r = history.query(session)
170 if r is not None: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 raise Reject(
172 "Signature for changes file was already seen at {0}.\nPlease refresh the signature of the changes file if you want to upload it again.".format(
173 r.seen
174 )
175 )
176 return True
178 @override
179 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
180 allow_source_untrusted_sig_keys = Config().value_list(
181 "Dinstall::AllowSourceUntrustedSigKeys"
182 )
184 changes = upload.changes
185 if not changes.valid_signature: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 raise Reject("Signature for .changes not valid.")
187 self.check_replay(upload)
188 self._check_hashes(upload, changes.filename, changes.files.values())
190 source = None
191 try:
192 source = changes.source
193 except Exception as e:
194 raise Reject("Invalid dsc file: {0}".format(e))
195 if source is not None:
196 if changes.primary_fingerprint not in allow_source_untrusted_sig_keys: 196 ↛ 201line 196 didn't jump to line 201 because the condition on line 196 was always true
197 if not source.valid_signature: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true
198 raise Reject("Signature for .dsc not valid.")
199 if source.primary_fingerprint != changes.primary_fingerprint: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 raise Reject(".changes and .dsc not signed by the same key.")
201 self._check_hashes(upload, source.filename, source.files.values())
203 if upload.fingerprint is None or upload.fingerprint.uid is None:
204 raise Reject(".changes signed by unknown key.")
206 return True
208 def _check_hashes(
209 self,
210 upload: "daklib.archive.ArchiveUpload",
211 filename: str,
212 files: Iterable[daklib.upload.HashedFile],
213 ) -> None:
214 """Make sure hashes match existing files
216 :param upload: upload we are processing
217 :param filename: name of the file the expected hash values are taken from
218 :param files: files to check the hashes for
219 """
220 try:
221 for f in files:
222 f.check(upload.directory)
223 except daklib.upload.FileDoesNotExist as e: 223 ↛ 231line 223 didn't jump to line 231
224 raise Reject(
225 "{0}: {1}\n"
226 "Perhaps you need to include the file in your upload?\n\n"
227 "If the orig tarball is missing, the -sa flag for dpkg-buildpackage will be your friend.".format(
228 filename, str(e)
229 )
230 )
231 except daklib.upload.UploadException as e:
232 raise Reject("{0}: {1}".format(filename, str(e)))
235class WeakSignatureCheck(Check):
236 """Check that .changes and .dsc are not signed using a weak algorithm"""
238 @override
239 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
240 changes = upload.changes
241 if changes.weak_signature: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 raise Reject(
243 "The .changes was signed using a weak algorithm (such as SHA-1)"
244 )
246 source = changes.source
247 if source is not None and source.weak_signature: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 raise Reject(
249 "The source package was signed using a weak algorithm (such as SHA-1)"
250 )
252 return True
255def check_signature_timestamp(prefix: str, signed_file: daklib.gpg.SignedFile) -> bool:
256 now = datetime.datetime.now(datetime.UTC)
257 timestamp = signed_file.signature_timestamp
258 age = now - timestamp
260 age_max = datetime.timedelta(days=365)
261 age_min = datetime.timedelta(days=-7)
263 if age > age_max: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 raise Reject(
265 "{0}: Signature from {1} is too old (maximum age is {2} days)".format(
266 prefix, timestamp, age_max.days
267 )
268 )
269 if age < age_min: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 raise Reject(
271 "{0}: Signature from {1} is too far in the future (tolerance is {2} days)".format(
272 prefix, timestamp, abs(age_min.days)
273 )
274 )
275 return True
278class SignatureTimestampCheck(Check):
279 """Check timestamp of .changes signature"""
281 @override
282 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
283 return check_signature_timestamp(
284 upload.changes.filename, upload.changes.signature
285 )
288class ChangesCheck(Check):
289 """Check changes file for syntax errors."""
291 @override
292 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
293 changes = upload.changes
294 control = changes.changes
295 fn = changes.filename
297 for field in (
298 "Distribution",
299 "Source",
300 "Architecture",
301 "Version",
302 "Maintainer",
303 "Files",
304 "Changes",
305 ):
306 if field not in control: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 raise Reject("{0}: misses mandatory field {1}".format(fn, field))
309 if len(changes.binaries) > 0:
310 for field in ("Binary", "Description"):
311 if field not in control: 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true
312 raise Reject(
313 "{0}: binary upload requires {1} field".format(fn, field)
314 )
316 check_fields_for_valid_utf8(fn, control)
318 source_match = re_field_source.match(control["Source"])
319 if not source_match: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 raise Reject("{0}: Invalid Source field".format(fn))
321 version_match = re_field_version.match(control["Version"])
322 if not version_match: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 raise Reject("{0}: Invalid Version field".format(fn))
324 version_without_epoch = version_match.group("without_epoch")
326 match = re_file_changes.match(fn)
327 if not match: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 raise Reject("{0}: Does not match re_file_changes".format(fn))
329 if match.group("package") != source_match.group("package"): 329 ↛ 330line 329 didn't jump to line 330 because the condition on line 329 was never true
330 raise Reject("{0}: Filename does not match Source field".format(fn))
331 if match.group("version") != version_without_epoch: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true
332 raise Reject("{0}: Filename does not match Version field".format(fn))
334 for bn in changes.binary_names:
335 if not re_field_package.match(bn): 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 raise Reject("{0}: Invalid binary package name {1}".format(fn, bn))
338 if changes.sourceful and changes.source is None: 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 raise Reject("Changes has architecture source, but no source found.")
340 if changes.source is not None and not changes.sourceful: 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 raise Reject("Upload includes source, but changes does not say so.")
343 try:
344 fix_maintainer(changes.changes["Maintainer"])
345 except ParseMaintError as e:
346 raise Reject(
347 "{0}: Failed to parse Maintainer field: {1}".format(changes.filename, e)
348 )
350 try:
351 changed_by = changes.changes.get("Changed-By")
352 if changed_by is not None: 352 ↛ 359line 352 didn't jump to line 359 because the condition on line 352 was always true
353 fix_maintainer(changed_by)
354 except ParseMaintError as e:
355 raise Reject(
356 "{0}: Failed to parse Changed-By field: {1}".format(changes.filename, e)
357 )
359 try:
360 changes.byhand_files
361 except daklib.upload.InvalidChangesException as e:
362 raise Reject("{0}".format(e))
364 if len(changes.files) == 0: 364 ↛ 365line 364 didn't jump to line 365 because the condition on line 364 was never true
365 raise Reject("Changes includes no files.")
367 for bugnum in changes.closed_bugs: 367 ↛ 368line 367 didn't jump to line 368 because the loop on line 367 never started
368 if not re_isanum.match(bugnum):
369 raise Reject(
370 '{0}: "{1}" in Closes field is not a number'.format(
371 changes.filename, bugnum
372 )
373 )
375 return True
378class ExternalHashesCheck(Check):
379 """Checks hashes in .changes and .dsc against an external database."""
381 def check_single(self, session: "Session", f):
382 q = session.execute(
383 sql.text(
384 "SELECT size, md5sum, sha1sum, sha256sum FROM external_files WHERE filename LIKE :pattern"
385 ),
386 {"pattern": "%/{}".format(f.filename)},
387 )
388 (ext_size, ext_md5sum, ext_sha1sum, ext_sha256sum) = q.fetchone() or (
389 None,
390 None,
391 None,
392 None,
393 )
395 if not ext_size:
396 return
398 if ext_size != f.size:
399 raise RejectExternalFilesMismatch(f.filename, "size", f.size, ext_size)
401 if ext_md5sum != f.md5sum:
402 raise RejectExternalFilesMismatch(
403 f.filename, "md5sum", f.md5sum, ext_md5sum
404 )
406 if ext_sha1sum != f.sha1sum:
407 raise RejectExternalFilesMismatch(
408 f.filename, "sha1sum", f.sha1sum, ext_sha1sum
409 )
411 if ext_sha256sum != f.sha256sum:
412 raise RejectExternalFilesMismatch(
413 f.filename, "sha256sum", f.sha256sum, ext_sha256sum
414 )
416 @override
417 def check(self, upload: "daklib.archive.ArchiveUpload"):
418 cnf = Config()
420 if not cnf.use_extfiles: # type: ignore[attr-defined] 420 ↛ 423line 420 didn't jump to line 423 because the condition on line 420 was always true
421 return
423 session = upload.session
424 changes = upload.changes
426 for f in changes.files.values():
427 self.check_single(session, f)
428 source = changes.source
429 if source is not None:
430 for f in source.files.values():
431 self.check_single(session, f)
434class BinaryCheck(Check):
435 """Check binary packages for syntax errors."""
437 @override
438 def check(self, upload):
439 debug_deb_name_postfix = "-dbgsym"
440 # XXX: Handle dynamic debug section name here
442 self._architectures: set[str] = set()
444 for binary in upload.changes.binaries:
445 self.check_binary(upload, binary)
447 for arch in upload.changes.architectures:
448 if arch == "source":
449 continue
450 if arch not in self._architectures: 450 ↛ 451line 450 didn't jump to line 451 because the condition on line 450 was never true
451 raise Reject(
452 "{}: Architecture field includes {}, but no binary packages for {} are included in the upload".format(
453 upload.changes.filename, arch, arch
454 )
455 )
457 binaries = {
458 binary.control["Package"]: binary for binary in upload.changes.binaries
459 }
461 for name, binary in list(binaries.items()):
462 if name in upload.changes.binary_names: 462 ↛ 465line 462 didn't jump to line 465 because the condition on line 462 was always true
463 # Package is listed in Binary field. Everything is good.
464 pass
465 elif daklib.utils.is_in_debug_section(binary.control):
466 # If we have a binary package in the debug section, we
467 # can allow it to not be present in the Binary field
468 # in the .changes file, so long as its name (without
469 # -dbgsym) is present in the Binary list.
470 if not name.endswith(debug_deb_name_postfix):
471 raise Reject(
472 "Package {0} is in the debug section, but "
473 "does not end in {1}.".format(name, debug_deb_name_postfix)
474 )
476 # Right, so, it's named properly, let's check that
477 # the corresponding package is in the Binary list
478 origin_package_name = name[: -len(debug_deb_name_postfix)]
479 if origin_package_name not in upload.changes.binary_names:
480 raise Reject(
481 "Debug package {debug}'s corresponding binary package "
482 "{origin} is not present in the Binary field.".format(
483 debug=name, origin=origin_package_name
484 )
485 )
486 else:
487 # Someone was a nasty little hacker and put a package
488 # into the .changes that isn't in debian/control. Bad,
489 # bad person.
490 raise Reject(
491 "Package {0} is not mentioned in Binary field in changes".format(
492 name
493 )
494 )
496 return True
498 def check_binary(self, upload: "daklib.archive.ArchiveUpload", binary):
499 fn = binary.hashed_file.filename
500 control = binary.control
502 for field in ("Package", "Architecture", "Version", "Description", "Section"):
503 if field not in control: 503 ↛ 504line 503 didn't jump to line 504 because the condition on line 503 was never true
504 raise Reject("{0}: Missing mandatory field {1}.".format(fn, field))
506 check_fields_for_valid_utf8(fn, control)
508 # check fields
510 package = control["Package"]
511 if not re_field_package.match(package): 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true
512 raise Reject("{0}: Invalid Package field".format(fn))
514 version = control["Version"]
515 version_match = re_field_version.match(version)
516 if not version_match: 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true
517 raise Reject("{0}: Invalid Version field".format(fn))
518 version_without_epoch = version_match.group("without_epoch")
520 architecture = control["Architecture"]
521 if architecture not in upload.changes.architectures: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 raise Reject(
523 "{0}: Architecture not in Architecture field in changes file".format(fn)
524 )
525 if architecture == "source": 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true
526 raise Reject(
527 '{0}: Architecture "source" invalid for binary packages'.format(fn)
528 )
529 self._architectures.add(architecture)
531 source = control.get("Source")
532 if source is not None and not re_field_source.match(source): 532 ↛ 533line 532 didn't jump to line 533 because the condition on line 532 was never true
533 raise Reject("{0}: Invalid Source field".format(fn))
535 section = control.get("Section", "")
536 if section == "" or section == "unknown" or section.endswith("/unknown"): 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 raise Reject(
538 '{0}: The "Section" field must be present and use a real section name.'.format(
539 fn
540 )
541 )
543 # check filename
545 match = re_file_binary.match(fn)
546 if match is None: 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true
547 raise Reject(f"{fn}: does not match re_file_binary")
548 if package != match.group("package"): 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true
549 raise Reject("{0}: filename does not match Package field".format(fn))
550 if version_without_epoch != match.group("version"): 550 ↛ 551line 550 didn't jump to line 551 because the condition on line 550 was never true
551 raise Reject("{0}: filename does not match Version field".format(fn))
552 if architecture != match.group("architecture"): 552 ↛ 553line 552 didn't jump to line 553 because the condition on line 552 was never true
553 raise Reject("{0}: filename does not match Architecture field".format(fn))
555 # check dependency field syntax
557 def check_dependency_field(
558 field,
559 control,
560 dependency_parser=apt_pkg.parse_depends,
561 allow_alternatives=True,
562 allow_relations=("", "<", "<=", "=", ">=", ">"),
563 ):
564 value = control.get(field)
565 if value is not None:
566 if value.strip() == "": 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 raise Reject("{0}: empty {1} field".format(fn, field))
568 try:
569 depends = dependency_parser(value)
570 except:
571 raise Reject("{0}: APT could not parse {1} field".format(fn, field))
572 for group in depends:
573 if not allow_alternatives and len(group) != 1: 573 ↛ 574line 573 didn't jump to line 574 because the condition on line 573 was never true
574 raise Reject(
575 "{0}: {1}: alternatives are not allowed".format(fn, field)
576 )
577 for dep_pkg, dep_ver, dep_rel in group:
578 if dep_rel not in allow_relations: 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true
579 raise Reject(
580 "{}: {}: depends on {}, but only relations {} are allowed for this field".format(
581 fn,
582 field,
583 " ".join([dep_pkg, dep_rel, dep_ver]),
584 allow_relations,
585 )
586 )
588 for field in (
589 "Breaks",
590 "Conflicts",
591 "Depends",
592 "Enhances",
593 "Pre-Depends",
594 "Recommends",
595 "Replaces",
596 "Suggests",
597 ):
598 check_dependency_field(field, control)
600 check_dependency_field(
601 "Provides", control, allow_alternatives=False, allow_relations=("", "=")
602 )
603 check_dependency_field(
604 "Built-Using",
605 control,
606 dependency_parser=apt_pkg.parse_src_depends,
607 allow_alternatives=False,
608 allow_relations=("=",),
609 )
612_DEB_ALLOWED_MEMBERS = {
613 "debian-binary",
614 *(f"control.tar.{comp}" for comp in ("gz", "xz")),
615 *(f"data.tar.{comp}" for comp in ("gz", "bz2", "xz")),
616}
619class BinaryMembersCheck(Check):
620 """check members of .deb file"""
622 @override
623 def check(self, upload: "daklib.archive.ArchiveUpload"):
624 for binary in upload.changes.binaries:
625 filename = binary.hashed_file.filename
626 path = os.path.join(upload.directory, filename)
627 self._check_binary(filename, path)
628 return True
630 def _check_binary(self, filename: str, path: str) -> None:
631 deb = apt_inst.DebFile(path)
632 members = {member.name for member in deb.getmembers()} # type: ignore[attr-defined]
633 if blocked_members := members - _DEB_ALLOWED_MEMBERS: 633 ↛ 634line 633 didn't jump to line 634 because the condition on line 633 was never true
634 raise Reject(
635 f"{filename}: Contains blocked members {', '.join(blocked_members)}"
636 )
639class BinaryTimestampCheck(Check):
640 """check timestamps of files in binary packages
642 Files in the near future cause ugly warnings and extreme time travel
643 can cause errors on extraction.
644 """
646 @override
647 def check(self, upload: "daklib.archive.ArchiveUpload"):
648 cnf = Config()
649 future_cutoff = time.time() + cnf.find_i(
650 "Dinstall::FutureTimeTravelGrace", 24 * 3600
651 )
652 past_cutoff = time.mktime(
653 time.strptime(cnf.find("Dinstall::PastCutoffYear", "1975"), "%Y")
654 )
656 class TarTime:
657 def __init__(self):
658 self.future_files: dict[str, int] = {}
659 self.past_files: dict[str, int] = {}
661 def callback(self, member, data) -> None:
662 if member.mtime > future_cutoff: 662 ↛ 663line 662 didn't jump to line 663 because the condition on line 662 was never true
663 self.future_files[member.name] = member.mtime
664 elif member.mtime < past_cutoff: 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true
665 self.past_files[member.name] = member.mtime
667 def format_reason(filename, direction, files) -> str:
668 reason = (
669 "{0}: has {1} file(s) with a timestamp too far in the {2}:\n".format(
670 filename, len(files), direction
671 )
672 )
673 for fn, ts in files.items():
674 reason += " {0} ({1})".format(fn, time.ctime(ts))
675 return reason
677 for binary in upload.changes.binaries:
678 filename = binary.hashed_file.filename
679 path = os.path.join(upload.directory, filename)
680 deb = apt_inst.DebFile(path)
681 tar = TarTime()
682 for archive in (deb.control, deb.data):
683 archive.go(tar.callback)
684 if tar.future_files: 684 ↛ 685line 684 didn't jump to line 685 because the condition on line 684 was never true
685 raise Reject(format_reason(filename, "future", tar.future_files))
686 if tar.past_files: 686 ↛ 687line 686 didn't jump to line 687 because the condition on line 686 was never true
687 raise Reject(format_reason(filename, "past", tar.past_files))
690class SourceCheck(Check):
691 """Check source package for syntax errors."""
693 def check_filename(self, control, filename, regex: re.Pattern) -> None:
694 # In case we have an .orig.tar.*, we have to strip the Debian revison
695 # from the version number. So handle this special case first.
696 is_orig = True
697 match = re_file_orig.match(filename)
698 if not match:
699 is_orig = False
700 match = regex.match(filename)
702 if not match: 702 ↛ 703line 702 didn't jump to line 703 because the condition on line 702 was never true
703 raise Reject(
704 "{0}: does not match regular expression for source filenames".format(
705 filename
706 )
707 )
708 if match.group("package") != control["Source"]: 708 ↛ 709line 708 didn't jump to line 709 because the condition on line 708 was never true
709 raise Reject("{0}: filename does not match Source field".format(filename))
711 version = control["Version"]
712 if is_orig:
713 upstream_match = re_field_version_upstream.match(version)
714 if not upstream_match: 714 ↛ 715line 714 didn't jump to line 715 because the condition on line 714 was never true
715 raise Reject(
716 "{0}: Source package includes upstream tarball, but {1} has no Debian revision.".format(
717 filename, version
718 )
719 )
720 version = upstream_match.group("upstream")
721 version_match = re_field_version.match(version)
722 if version_match is None: 722 ↛ 723line 722 didn't jump to line 723 because the condition on line 722 was never true
723 raise Reject(f"{filename}: Version field does not match re_field_version")
724 version_without_epoch = version_match.group("without_epoch")
725 if match.group("version") != version_without_epoch: 725 ↛ 726line 725 didn't jump to line 726 because the condition on line 725 was never true
726 raise Reject("{0}: filename does not match Version field".format(filename))
728 @override
729 def check(self, upload: "daklib.archive.ArchiveUpload"):
730 if upload.changes.source is None:
731 if upload.changes.sourceful: 731 ↛ 732line 731 didn't jump to line 732 because the condition on line 731 was never true
732 raise Reject(
733 "{}: Architecture field includes source, but no source package is included in the upload".format(
734 upload.changes.filename
735 )
736 )
737 return True
739 if not upload.changes.sourceful: 739 ↛ 740line 739 didn't jump to line 740 because the condition on line 739 was never true
740 raise Reject(
741 "{}: Architecture field does not include source, but a source package is included in the upload".format(
742 upload.changes.filename
743 )
744 )
746 changes = upload.changes.changes
747 source = upload.changes.source
748 control = cast(apt_pkg.TagSection, source.dsc)
749 dsc_fn = source._dsc_file.filename
751 check_fields_for_valid_utf8(dsc_fn, control)
753 # check fields
754 if not re_field_package.match(control["Source"]): 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 raise Reject("{0}: Invalid Source field".format(dsc_fn))
756 if control["Source"] != changes["Source"]: 756 ↛ 757line 756 didn't jump to line 757 because the condition on line 756 was never true
757 raise Reject(
758 "{0}: Source field does not match Source field in changes".format(
759 dsc_fn
760 )
761 )
762 if control["Version"] != changes["Version"]: 762 ↛ 763line 762 didn't jump to line 763 because the condition on line 762 was never true
763 raise Reject(
764 "{0}: Version field does not match Version field in changes".format(
765 dsc_fn
766 )
767 )
769 # check filenames
770 self.check_filename(control, dsc_fn, re_file_dsc)
771 for f in source.files.values():
772 self.check_filename(control, f.filename, re_file_source)
774 # check dependency field syntax
775 for field in (
776 "Build-Conflicts",
777 "Build-Conflicts-Indep",
778 "Build-Depends",
779 "Build-Depends-Arch",
780 "Build-Depends-Indep",
781 ):
782 value = control.get(field)
783 if value is not None:
784 if value.strip() == "": 784 ↛ 785line 784 didn't jump to line 785 because the condition on line 784 was never true
785 raise Reject("{0}: empty {1} field".format(dsc_fn, field))
786 try:
787 apt_pkg.parse_src_depends(value)
788 except Exception as e:
789 raise Reject(
790 "{0}: APT could not parse {1} field: {2}".format(
791 dsc_fn, field, e
792 )
793 )
795 rejects = utils.check_dsc_files(dsc_fn, control, list(source.files.keys()))
796 if len(rejects) > 0: 796 ↛ 797line 796 didn't jump to line 797 because the condition on line 796 was never true
797 raise Reject("\n".join(rejects))
799 return True
802class SingleDistributionCheck(Check):
803 """Check that the .changes targets only a single distribution."""
805 @override
806 def check(self, upload):
807 if len(upload.changes.distributions) != 1: 807 ↛ 808line 807 didn't jump to line 808 because the condition on line 807 was never true
808 raise Reject("Only uploads to a single distribution are allowed.")
811class ACLCheck(Check):
812 """Check the uploader is allowed to upload the packages in .changes"""
814 def _does_hijack(
815 self, session: "Session", upload: "daklib.archive.ArchiveUpload", suite: Suite
816 ) -> tuple[Literal[True], str, str] | tuple[Literal[False], None, None]:
817 # Try to catch hijacks.
818 # This doesn't work correctly. Uploads to experimental can still
819 # "hijack" binaries from unstable. Also one can hijack packages
820 # via buildds (but people who try this should not be DMs).
821 for binary_name in upload.changes.binary_names:
822 binaries_query = (
823 select(DBBinary)
824 .join(DBBinary.source)
825 .where(DBBinary.suites.contains(suite))
826 .where(DBBinary.package == binary_name)
827 )
828 for binary in session.scalars(binaries_query):
829 if binary.source.source != upload.changes.changes["Source"]: 829 ↛ 830line 829 didn't jump to line 830 because the condition on line 829 was never true
830 return True, binary.package, binary.source.source
831 return False, None, None
833 def _check_acl(
834 self, session: "Session", upload: "daklib.archive.ArchiveUpload", acl: ACL
835 ) -> tuple[Literal[False] | None, str] | tuple[Literal[True], None]:
836 source_name = upload.changes.source_name
837 fingerprint = upload.authorized_by_fingerprint
839 if acl.match_fingerprint and fingerprint not in acl.fingerprints: 839 ↛ 840line 839 didn't jump to line 840 because the condition on line 839 was never true
840 return None, "Fingerprint not in ACL"
841 if acl.match_keyring is not None and fingerprint.keyring != acl.match_keyring: 841 ↛ 842line 841 didn't jump to line 842 because the condition on line 841 was never true
842 return None, "Fingerprint not in ACL's keyring"
844 if not acl.allow_new:
845 if upload.new:
846 return False, "NEW uploads are not allowed"
847 for f in upload.changes.files.values():
848 assert f.section is not None
849 if f.section == "byhand" or f.section.startswith("raw-"): 849 ↛ 850line 849 didn't jump to line 850 because the condition on line 849 was never true
850 return False, "BYHAND uploads are not allowed"
851 if not acl.allow_source and upload.changes.source is not None: 851 ↛ 852line 851 didn't jump to line 852 because the condition on line 851 was never true
852 return False, "sourceful uploads are not allowed"
853 binaries = upload.changes.binaries
854 if len(binaries) != 0:
855 if not acl.allow_binary: 855 ↛ 856line 855 didn't jump to line 856 because the condition on line 855 was never true
856 return False, "binary uploads are not allowed"
857 if upload.changes.source is None and not acl.allow_binary_only: 857 ↛ 858line 857 didn't jump to line 858 because the condition on line 857 was never true
858 return False, "binary-only uploads are not allowed"
859 if not acl.allow_binary_all: 859 ↛ 860line 859 didn't jump to line 860 because the condition on line 859 was never true
860 uploaded_arches = set(upload.changes.architectures)
861 uploaded_arches.discard("source")
862 allowed_arches = {a.arch_string for a in acl.architectures}
863 forbidden_arches = uploaded_arches - allowed_arches
864 if len(forbidden_arches) != 0:
865 return (
866 False,
867 "uploads for architecture(s) {0} are not allowed".format(
868 ", ".join(forbidden_arches)
869 ),
870 )
871 if not acl.allow_hijack:
872 assert upload.final_suites is not None
873 for suite in upload.final_suites:
874 does_hijack, hijacked_binary, hijacked_from = self._does_hijack(
875 session, upload, suite
876 )
877 if does_hijack: 877 ↛ 878line 877 didn't jump to line 878 because the condition on line 877 was never true
878 return (
879 False,
880 "hijacks are not allowed (binary={0}, other-source={1})".format(
881 hijacked_binary, hijacked_from
882 ),
883 )
885 acl_per_source = session.scalars(
886 select(ACLPerSource)
887 .filter_by(acl=acl, fingerprint=fingerprint, source=source_name)
888 .limit(1)
889 ).first()
890 if acl.allow_per_source and acl_per_source is None:
891 return False, "not allowed to upload source package '{0}'".format(
892 source_name
893 )
894 if acl.deny_per_source and acl_per_source is not None: 894 ↛ 895line 894 didn't jump to line 895 because the condition on line 894 was never true
895 return (
896 False,
897 acl_per_source.reason
898 or "forbidden to upload source package '{0}'".format(source_name),
899 )
901 return True, None
903 @override
904 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
905 session = upload.session
906 fingerprint = upload.authorized_by_fingerprint
907 keyring = fingerprint.keyring
909 if keyring is None: 909 ↛ 910line 909 didn't jump to line 910 because the condition on line 909 was never true
910 raise Reject(
911 "No keyring for fingerprint {0}".format(fingerprint.fingerprint)
912 )
913 if not keyring.active: 913 ↛ 914line 913 didn't jump to line 914 because the condition on line 913 was never true
914 raise Reject("Keyring {0} is not active".format(keyring.keyring_name))
916 acl = fingerprint.acl or keyring.acl
917 if acl is None: 917 ↛ 918line 917 didn't jump to line 918 because the condition on line 917 was never true
918 raise Reject("No ACL for fingerprint {0}".format(fingerprint.fingerprint))
919 result, reason = self._check_acl(session, upload, acl)
920 if not result:
921 assert reason is not None
922 raise RejectACL(acl, reason)
924 for acl in session.scalars(select(ACL).filter_by(is_global=True)):
925 result, reason = self._check_acl(session, upload, acl)
926 if result is False: 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 assert reason is not None
928 raise RejectACL(acl, reason)
930 return True
932 @override
933 def per_suite_check(
934 self, upload: "daklib.archive.ArchiveUpload", suite: Suite
935 ) -> bool:
936 acls = suite.acls
937 if len(acls) != 0: 937 ↛ 938line 937 didn't jump to line 938 because the condition on line 937 was never true
938 accept = False
939 for acl in acls:
940 result, reason = self._check_acl(upload.session, upload, acl)
941 if result is False:
942 raise Reject(reason)
943 accept = accept or bool(result)
944 if not accept:
945 raise Reject(
946 "Not accepted by any per-suite acl (suite={0})".format(
947 suite.suite_name
948 )
949 )
950 return True
953class TransitionCheck(Check):
954 """check for a transition"""
956 @override
957 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
958 if not upload.changes.sourceful:
959 return True
961 transitions = self.get_transitions()
962 if transitions is None: 962 ↛ 965line 962 didn't jump to line 965 because the condition on line 962 was always true
963 return True
965 session = upload.session
967 control = upload.changes.changes
968 source_match = re_field_source.match(control["Source"])
969 if source_match is None:
970 raise Reject(
971 f"{upload.changes.filename}: Source field does not match re_field_source"
972 )
973 source = source_match.group("package")
975 for trans in transitions:
976 t = transitions[trans]
977 transition_source = t["source"]
978 expected = t["new"]
980 # Will be None if nothing is in testing.
981 current = get_source_in_suite(transition_source, "testing", session)
982 if current is not None:
983 compare = apt_pkg.version_compare(current.version, expected)
985 # This is still valid, the current version in testing is older than
986 # the new version we wait for, or there is none in testing yet.
987 # Check if the source we look at is affected by this.
988 if current is None or compare < 0 and source in t["packages"]:
989 # The source is affected, lets reject it.
991 rejectmsg = "{0}: part of the {1} transition.\n\n".format(source, trans)
993 if current is not None:
994 currentlymsg = "at version {0}".format(current.version)
995 else:
996 currentlymsg = "not present in testing"
998 rejectmsg += "Transition description: {0}\n\n".format(t["reason"])
1000 rejectmsg += "\n".join(
1001 textwrap.wrap(
1002 """Your package
1003is part of a testing transition designed to get {0} migrated (it is
1004currently {1}, we need version {2}). This transition is managed by the
1005Release Team, and {3} is the Release-Team member responsible for it.
1006Please mail debian-release@lists.debian.org or contact {3} directly if you
1007need further assistance. You might want to upload to experimental until this
1008transition is done.""".format(
1009 transition_source, currentlymsg, expected, t["rm"]
1010 )
1011 )
1012 )
1014 raise Reject(rejectmsg)
1016 return True
1018 def get_transitions(self):
1019 cnf = Config()
1020 path = cnf.get("Dinstall::ReleaseTransitions", "")
1021 if path == "" or not os.path.exists(path): 1021 ↛ 1024line 1021 didn't jump to line 1024 because the condition on line 1021 was always true
1022 return None
1024 with open(path, "r") as fd:
1025 contents = fd.read()
1026 try:
1027 transitions = yaml.safe_load(contents)
1028 return transitions
1029 except yaml.YAMLError as msg:
1030 utils.warn(
1031 "Not checking transitions, the transitions file is broken: {0}".format(
1032 msg
1033 )
1034 )
1036 return None
1039class NoSourceOnlyCheck(Check):
1040 def is_source_only_upload(self, upload: "daklib.archive.ArchiveUpload") -> bool:
1041 changes = upload.changes
1042 return bool(changes.source is not None and len(changes.binaries) == 0)
1044 """Check for source-only upload
1046 Source-only uploads are only allowed if Dinstall::AllowSourceOnlyUploads is
1047 set. Otherwise they are rejected.
1049 Source-only uploads are only accepted for source packages having a
1050 Package-List field that also lists architectures per package. This
1051 check can be disabled via
1052 Dinstall::AllowSourceOnlyUploadsWithoutPackageList.
1054 Source-only uploads to NEW are only allowed if
1055 Dinstall::AllowSourceOnlyNew is set.
1057 Uploads not including architecture-independent packages are only
1058 allowed if Dinstall::AllowNoArchIndepUploads is set.
1060 """
1062 @override
1063 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
1064 if not self.is_source_only_upload(upload):
1065 return True
1066 changes = upload.changes
1067 assert changes.source is not None
1069 allow_source_only_uploads = Config().find_b("Dinstall::AllowSourceOnlyUploads")
1070 allow_source_only_uploads_without_package_list = Config().find_b(
1071 "Dinstall::AllowSourceOnlyUploadsWithoutPackageList"
1072 )
1073 allow_source_only_new = Config().find_b("Dinstall::AllowSourceOnlyNew")
1074 allow_source_only_new_keys = Config().value_list(
1075 "Dinstall::AllowSourceOnlyNewKeys"
1076 )
1077 allow_source_only_new_sources = Config().value_list(
1078 "Dinstall::AllowSourceOnlyNewSources"
1079 )
1080 allow_no_arch_indep_uploads = Config().find_b(
1081 "Dinstall::AllowNoArchIndepUploads", True
1082 )
1084 if not allow_source_only_uploads: 1084 ↛ 1085line 1084 didn't jump to line 1085 because the condition on line 1084 was never true
1085 raise Reject("Source-only uploads are not allowed.")
1086 if ( 1086 ↛ 1090line 1086 didn't jump to line 1090
1087 not allow_source_only_uploads_without_package_list
1088 and changes.source.package_list.fallback
1089 ):
1090 raise Reject(
1091 "Source-only uploads are only allowed if a Package-List field that also list architectures is included in the source package. dpkg (>= 1.17.7) includes this information."
1092 )
1093 if ( 1093 ↛ 1099line 1093 didn't jump to line 1099
1094 not allow_source_only_new
1095 and upload.new
1096 and changes.primary_fingerprint not in allow_source_only_new_keys
1097 and changes.source_name not in allow_source_only_new_sources
1098 ):
1099 raise Reject("Source-only uploads to NEW are not allowed.")
1101 if ( 1101 ↛ 1106line 1101 didn't jump to line 1106
1102 "all" not in changes.architectures
1103 and changes.source.package_list.has_arch_indep_packages()
1104 and not allow_no_arch_indep_uploads
1105 ):
1106 raise Reject("Uploads must include architecture-independent packages.")
1108 return True
1111class NewOverrideCheck(Check):
1112 """Override NEW requirement"""
1114 @override
1115 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
1116 if not upload.new:
1117 return True
1119 new_override_keys = Config().value_list("Dinstall::NewOverrideKeys")
1120 changes = upload.changes
1122 if changes.primary_fingerprint in new_override_keys: 1122 ↛ 1123line 1122 didn't jump to line 1123 because the condition on line 1122 was never true
1123 upload.new = False
1125 return True
1128class ArchAllBinNMUCheck(Check):
1129 """Check for arch:all binNMUs"""
1131 @override
1132 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
1133 changes = upload.changes
1135 if ( 1135 ↛ 1139line 1135 didn't jump to line 1139
1136 "all" in changes.architectures
1137 and changes.changes.get("Binary-Only") == "yes"
1138 ):
1139 raise Reject("arch:all binNMUs are not allowed.")
1141 return True
1144class LintianCheck(Check):
1145 """Check package using lintian"""
1147 @override
1148 def check(self, upload: "daklib.archive.ArchiveUpload") -> bool:
1149 changes = upload.changes
1151 # Only check sourceful uploads.
1152 if changes.source is None:
1153 return True
1154 # Only check uploads to unstable or experimental.
1155 if ( 1155 ↛ 1159line 1155 didn't jump to line 1159
1156 "unstable" not in changes.distributions
1157 and "experimental" not in changes.distributions
1158 ):
1159 return True
1161 cnf = Config()
1162 if "Dinstall::LintianTags" not in cnf:
1163 return True
1164 tagfile = cnf["Dinstall::LintianTags"]
1166 with open(tagfile, "r") as sourcefile:
1167 sourcecontent = sourcefile.read()
1168 try:
1169 lintiantags = yaml.safe_load(sourcecontent)["lintian"]
1170 except yaml.YAMLError as msg:
1171 raise Exception(
1172 "Could not read lintian tags file {0}, YAML error: {1}".format(
1173 tagfile, msg
1174 )
1175 )
1177 with tempfile.NamedTemporaryFile(mode="w+t") as temptagfile:
1178 os.fchmod(temptagfile.fileno(), 0o644)
1179 for tags in lintiantags.values():
1180 for tag in tags:
1181 print(tag, file=temptagfile)
1182 temptagfile.flush()
1184 changespath = os.path.join(upload.directory, changes.filename)
1186 tempdir = cnf.get("Dir::TempPath") or os.environ.get("TMPDIR", "/tmp")
1187 cmd = []
1188 user = cnf.get("Dinstall::UnprivUser") or None
1189 if user is not None: 1189 ↛ 1190line 1189 didn't jump to line 1190 because the condition on line 1189 was never true
1190 cmd.extend(["sudo", "-H", "-u", user, "TMPDIR={0}".format(tempdir)])
1191 cmd.extend(
1192 [
1193 "/usr/bin/lintian",
1194 "--show-overrides",
1195 "--tags-from-file",
1196 temptagfile.name,
1197 changespath,
1198 ]
1199 )
1200 process = subprocess.run(
1201 cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
1202 )
1203 output = process.stdout
1204 result = process.returncode
1206 if result == 2: 1206 ↛ 1207line 1206 didn't jump to line 1207 because the condition on line 1206 was never true
1207 utils.warn(
1208 "lintian failed for %s [return code: %s]." % (changespath, result)
1209 )
1210 utils.warn(utils.prefix_multi_line_string(output, " [possible output:] "))
1212 parsed_tags = lintian.parse_lintian_output(output)
1213 rejects = list(lintian.generate_reject_messages(parsed_tags, lintiantags))
1214 if len(rejects) != 0: 1214 ↛ 1215line 1214 didn't jump to line 1215 because the condition on line 1214 was never true
1215 raise Reject("\n".join(rejects))
1217 return True
1220class SourceFormatCheck(Check):
1221 """Check source format is allowed in the target suite"""
1223 @override
1224 def per_suite_check(
1225 self, upload: "daklib.archive.ArchiveUpload", suite: Suite
1226 ) -> bool:
1227 source = upload.changes.source
1228 session = upload.session
1229 if source is None:
1230 return True
1232 source_format = source.dsc["Format"]
1233 query = (
1234 select(SrcFormat)
1235 .where(SrcFormat.format_name == source_format)
1236 .where(SrcFormat.suites.contains(suite))
1237 .limit(1)
1238 )
1239 if session.scalars(query).first() is None:
1240 raise Reject(
1241 "source format {0} is not allowed in suite {1}".format(
1242 source_format, suite.suite_name
1243 )
1244 )
1246 return True
1249class SuiteCheck(Check):
1250 @override
1251 def per_suite_check(
1252 self, upload: "daklib.archive.ArchiveUpload", suite: Suite
1253 ) -> bool:
1254 if not suite.accept_source_uploads and upload.changes.source is not None: 1254 ↛ 1255line 1254 didn't jump to line 1255 because the condition on line 1254 was never true
1255 raise Reject(
1256 'The suite "{0}" does not accept source uploads.'.format(
1257 suite.suite_name
1258 )
1259 )
1260 if not suite.accept_binary_uploads and len(upload.changes.binaries) != 0: 1260 ↛ 1261line 1260 didn't jump to line 1261 because the condition on line 1260 was never true
1261 raise Reject(
1262 'The suite "{0}" does not accept binary uploads.'.format(
1263 suite.suite_name
1264 )
1265 )
1266 return True
1269class SuiteArchitectureCheck(Check):
1270 @override
1271 def per_suite_check(
1272 self, upload: "daklib.archive.ArchiveUpload", suite: Suite
1273 ) -> bool:
1274 session = upload.session
1275 for arch in upload.changes.architectures:
1276 query = (
1277 select(Architecture)
1278 .where(Architecture.arch_string == arch)
1279 .where(Architecture.suites.contains(suite))
1280 .limit(1)
1281 )
1282 if session.scalars(query).first() is None:
1283 raise Reject(
1284 "Architecture {0} is not allowed in suite {1}".format(
1285 arch, suite.suite_name
1286 )
1287 )
1289 return True
1292class VersionCheck(Check):
1293 """Check version constraints"""
1295 def _highest_source_version(
1296 self, session: "Session", source_name: str, suite: Suite
1297 ) -> str | None:
1298 db_source = session.scalars(
1299 select(DBSource)
1300 .where(DBSource.source == source_name)
1301 .where(DBSource.suites.contains(suite))
1302 .order_by(DBSource.version.desc())
1303 .limit(1)
1304 ).first()
1305 if db_source is None:
1306 return None
1307 else:
1308 return db_source.version
1310 def _highest_binary_version(
1311 self, session: "Session", binary_name: str, suite: Suite, architecture: str
1312 ) -> str | None:
1313 db_binary = session.scalars(
1314 select(DBBinary)
1315 .where(DBBinary.package == binary_name)
1316 .where(DBBinary.suites.contains(suite))
1317 .join(DBBinary.architecture)
1318 .where(Architecture.arch_string.in_(["all", architecture]))
1319 .order_by(DBBinary.version.desc())
1320 .limit(1)
1321 ).first()
1322 if db_binary is None:
1323 return None
1324 else:
1325 return db_binary.version
1327 def _version_checks(
1328 self,
1329 upload: "daklib.archive.ArchiveUpload",
1330 suite: Suite,
1331 other_suite: Suite,
1332 op: Callable[[int], bool],
1333 op_name: str,
1334 ) -> None:
1335 session = upload.session
1337 if upload.changes.source is not None:
1338 source_name = upload.changes.source.dsc["Source"]
1339 source_version = upload.changes.source.dsc["Version"]
1340 v = self._highest_source_version(session, source_name, other_suite)
1341 if v is not None and not op(version_compare(source_version, v)): 1341 ↛ 1342line 1341 didn't jump to line 1342 because the condition on line 1341 was never true
1342 raise Reject(
1343 "Version check failed:\n"
1344 "Your upload included the source package {0}, version {1},\n"
1345 "however {3} already has version {2}.\n"
1346 "Uploads to {5} must have a {4} version than present in {3}.".format(
1347 source_name,
1348 source_version,
1349 v,
1350 other_suite.suite_name,
1351 op_name,
1352 suite.suite_name,
1353 )
1354 )
1356 for binary in upload.changes.binaries:
1357 binary_name = binary.control["Package"]
1358 binary_version = binary.control["Version"]
1359 architecture = binary.control["Architecture"]
1360 v = self._highest_binary_version(
1361 session, binary_name, other_suite, architecture
1362 )
1363 if v is not None and not op(version_compare(binary_version, v)): 1363 ↛ 1364line 1363 didn't jump to line 1364 because the condition on line 1363 was never true
1364 raise Reject(
1365 "Version check failed:\n"
1366 "Your upload included the binary package {0}, version {1}, for {2},\n"
1367 "however {4} already has version {3}.\n"
1368 "Uploads to {6} must have a {5} version than present in {4}.".format(
1369 binary_name,
1370 binary_version,
1371 architecture,
1372 v,
1373 other_suite.suite_name,
1374 op_name,
1375 suite.suite_name,
1376 )
1377 )
1379 @override
1380 def per_suite_check(
1381 self, upload: "daklib.archive.ArchiveUpload", suite: Suite
1382 ) -> bool:
1383 session = upload.session
1385 vc_newer = session.scalars(
1386 select(dbconn.VersionCheck)
1387 .filter_by(suite=suite)
1388 .where(dbconn.VersionCheck.check.in_(["MustBeNewerThan", "Enhances"]))
1389 )
1390 must_be_newer_than = [vc.reference for vc in vc_newer]
1391 # Must be newer than old versions in `suite`
1392 must_be_newer_than.append(suite)
1394 for s in must_be_newer_than:
1395 self._version_checks(upload, suite, s, lambda result: result > 0, "higher")
1397 vc_older = session.scalars(
1398 select(dbconn.VersionCheck).filter_by(suite=suite, check="MustBeOlderThan")
1399 )
1400 must_be_older_than = [vc.reference for vc in vc_older]
1402 for s in must_be_older_than: 1402 ↛ 1403line 1402 didn't jump to line 1403 because the loop on line 1402 never started
1403 self._version_checks(upload, suite, s, lambda result: result < 0, "lower")
1405 return True
1407 @property
1408 @override
1409 def forcable(self) -> bool:
1410 return True