Coverage for daklib/command.py: 65%
267 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1"""module to handle command files
3@contact: Debian FTP Master <ftpmaster@debian.org>
4@copyright: 2012, Ansgar Burchardt <ansgar@debian.org>
5@copyright: 2023 Emilio Pozuelo Monfort <pochu@debian.org>
6@license: GPL-2+
7"""
9# This program is free software; you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation; either version 2 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License along
20# with this program; if not, write to the Free Software Foundation, Inc.,
21# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
23import os
24import tempfile
25from typing import TYPE_CHECKING, cast
27import apt_pkg
28from sqlalchemy import delete, select
29from sqlalchemy.engine import CursorResult
31from daklib.config import Config
32from daklib.dak_exceptions import ParseMaintError
33from daklib.dbconn import (
34 ACL,
35 ACLPerSource,
36 ACLPerSuite,
37 DBChange,
38 DBConn,
39 DBSource,
40 Fingerprint,
41 PolicyQueueUpload,
42 SignatureHistory,
43 get_active_keyring_files,
44)
45from daklib.gpg import SignedFile
46from daklib.regexes import re_field_package
47from daklib.textutils import fix_maintainer
48from daklib.utils import TemplateSubst, gpg_get_key_addresses, send_mail
50if TYPE_CHECKING:
51 from sqlalchemy.orm import Session
54class CommandError(Exception):
55 pass
58class CommandFile:
59 def __init__(self, filename: str, data: bytes, log=None):
60 if log is None: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 from daklib.daklog import Logger
63 log = Logger()
64 self.cc: list[str] = []
65 self.result: list[str] = []
66 self.log = log
67 self.filename: str = filename
68 self.data = data
69 self.uploader: str | None = None
71 def _check_replay(self, signed_file: SignedFile, session: "Session"):
72 """check for replays
74 .. note::
76 Will commit changes to the database.
78 :param session: database session
79 """
80 # Mark commands file as seen to prevent replays.
81 signature_history = SignatureHistory.from_signed_file(signed_file)
82 session.add(signature_history)
83 session.commit()
85 def _quote_section(self, section: apt_pkg.TagSection) -> str:
86 lines = [f"> {line}" for line in str(section).splitlines()]
87 return "\n".join(lines)
89 def _evaluate_sections(self, sections: apt_pkg.TagFile, session: "Session"):
90 session.rollback()
91 try:
92 while True:
93 next(sections)
94 section: apt_pkg.TagSection = sections.section # type: ignore[attr-defined]
95 self.result.append(self._quote_section(section))
97 action = section.get("Action", None)
98 if action is None: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 raise CommandError("Encountered section without Action field")
101 if action == "dm":
102 self.action_dm(self.fingerprint, section, session)
103 elif action == "dm-remove": 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true
104 self.action_dm_remove(self.fingerprint, section, session)
105 elif action == "dm-migrate": 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 self.action_dm_migrate(self.fingerprint, section, session)
107 elif action == "break-the-archive": 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 self.action_break_the_archive(self.fingerprint, section, session)
109 elif action == "process-upload": 109 ↛ 112line 109 didn't jump to line 112 because the condition on line 109 was always true
110 self.action_process_upload(self.fingerprint, section, session)
111 else:
112 raise CommandError("Unknown action: {0}".format(action))
114 self.result.append("")
115 except StopIteration:
116 pass
117 finally:
118 session.rollback()
120 def _notify_uploader(self):
121 cnf = Config()
123 bcc = "X-DAK: dak process-command"
124 if "Dinstall::Bcc" in cnf: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 bcc = "{0}\nBcc: {1}".format(bcc, cnf["Dinstall::Bcc"])
127 maint_to = ""
128 addresses = gpg_get_key_addresses(self.fingerprint.fingerprint)
129 if len(addresses) > 0: 129 ↛ 132line 129 didn't jump to line 132 because the condition on line 129 was always true
130 maint_to = addresses[0]
132 if self.uploader:
133 try:
134 maint_to = fix_maintainer(self.uploader)[1]
135 except ParseMaintError:
136 self.log.log("ignoring malformed uploader", self.filename)
138 cc = set()
139 for address in self.cc:
140 try:
141 cc.add(fix_maintainer(address)[1])
142 except ParseMaintError:
143 self.log.log("ignoring malformed cc", self.filename)
145 subst = {
146 "__DAK_ADDRESS__": cnf["Dinstall::MyEmailAddress"],
147 "__MAINTAINER_TO__": maint_to,
148 "__CC__": ", ".join(cc),
149 "__BCC__": bcc,
150 "__RESULTS__": "\n".join(self.result),
151 "__FILENAME__": self.filename,
152 }
154 message = TemplateSubst(
155 subst, os.path.join(cnf["Dir::Templates"], "process-command.processed")
156 )
158 send_mail(message)
160 def evaluate(self) -> bool:
161 """evaluate commands file
163 :return: :const:`True` if the file was processed sucessfully,
164 :const:`False` otherwise
165 """
166 result = True
168 session = DBConn().session()
170 keyring_files = get_active_keyring_files(session)
172 signed_file = SignedFile(self.data, keyring_files)
173 if not signed_file.valid: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 self.log.log(["invalid signature", self.filename])
175 return False
177 self.fingerprint = session.execute(
178 select(Fingerprint).filter_by(fingerprint=signed_file.primary_fingerprint)
179 ).scalar_one()
180 if self.fingerprint.keyring is None: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 self.log.log(["signed by key in unknown keyring", self.filename])
182 return False
183 assert self.fingerprint.keyring.active
185 self.log.log(
186 [
187 "processing",
188 self.filename,
189 "signed-by={0}".format(self.fingerprint.fingerprint),
190 ]
191 )
193 with tempfile.TemporaryFile() as fh:
194 fh.write(signed_file.contents)
195 fh.seek(0)
196 sections = apt_pkg.TagFile(fh)
198 try:
199 next(sections)
200 section: apt_pkg.TagSection = sections.section # type: ignore[attr-defined]
201 if "Uploader" in section:
202 self.uploader = section["Uploader"]
203 if "Cc" in section: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 self.cc.append(section["Cc"])
205 # TODO: Verify first section has valid Archive field
206 if "Archive" not in section: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 raise CommandError("No Archive field in first section.")
209 # TODO: send mail when we detected a replay.
210 self._check_replay(signed_file, session)
212 self._evaluate_sections(sections, session)
213 self.result.append("")
214 except Exception as e:
215 self.log.log(["ERROR", e])
216 self.result.append(
217 "There was an error processing this section. No changes were committed.\nDetails:\n{0}".format(
218 e
219 )
220 )
221 result = False
223 self._notify_uploader()
225 session.close()
227 return result
229 def _split_packages(self, value: str) -> list[str]:
230 names = value.split()
231 for name in names:
232 if not re_field_package.match(name): 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 raise CommandError('Invalid package name "{0}"'.format(name))
234 return names
236 def action_dm(
237 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
238 ) -> None:
239 cnf = Config()
241 if ( 241 ↛ 246line 241 didn't jump to line 246
242 "Command::DM::AdminKeyrings" not in cnf
243 or "Command::DM::ACL" not in cnf
244 or "Command::DM::Keyrings" not in cnf
245 ):
246 raise CommandError("DM command is not configured for this archive.")
248 allowed_keyrings = cnf.value_list("Command::DM::AdminKeyrings")
249 if ( 249 ↛ 253line 249 didn't jump to line 253
250 fingerprint.keyring is None
251 or fingerprint.keyring.keyring_name not in allowed_keyrings
252 ):
253 raise CommandError(
254 "Key {0} is not allowed to set DM".format(fingerprint.fingerprint)
255 )
257 acl_name = cnf.get("Command::DM::ACL", "dm")
258 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one()
260 fpr_hash = section["Fingerprint"].replace(" ", "")
261 fpr = session.scalars(
262 select(Fingerprint).filter_by(fingerprint=fpr_hash).limit(1)
263 ).first()
264 if fpr is None: 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 raise CommandError("Unknown fingerprint {0}".format(fpr_hash))
266 if fpr.keyring is None or fpr.keyring.keyring_name not in cnf.value_list(
267 "Command::DM::Keyrings"
268 ):
269 raise CommandError("Key {0} is not in DM keyring.".format(fpr.fingerprint))
270 addresses = gpg_get_key_addresses(fpr.fingerprint)
271 if len(addresses) > 0: 271 ↛ 274line 271 didn't jump to line 274 because the condition on line 271 was always true
272 self.cc.append(addresses[0])
274 self.log.log(["dm", "fingerprint", fpr.fingerprint])
275 self.result.append("Fingerprint: {0}".format(fpr.fingerprint))
276 if len(addresses) > 0: 276 ↛ 280line 276 didn't jump to line 280 because the condition on line 276 was always true
277 self.log.log(["dm", "uid", addresses[0]])
278 self.result.append("Uid: {0}".format(addresses[0]))
280 for source in self._split_packages(section.get("Allow", "")):
281 # Check for existance of source package to catch typos
282 source_query = select(DBSource).filter_by(source=source).limit(1)
283 if session.scalars(source_query).first() is None: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 raise CommandError(
285 "Tried to grant permissions for unknown source package: {0}".format(
286 source
287 )
288 )
290 if ( 290 ↛ 308line 290 didn't jump to line 308
291 session.scalars(
292 select(ACLPerSource)
293 .filter_by(acl=acl, fingerprint=fpr, source=source)
294 .limit(1)
295 ).first()
296 is None
297 ):
298 aps = ACLPerSource()
299 aps.acl = acl
300 aps.fingerprint = fpr
301 aps.source = source
302 aps.created_by = fingerprint
303 aps.reason = section.get("Reason")
304 session.add(aps)
305 self.log.log(["dm", "allow", fpr.fingerprint, source])
306 self.result.append("Allowed: {0}".format(source))
307 else:
308 self.result.append("Already-Allowed: {0}".format(source))
310 session.flush()
312 for source in self._split_packages(section.get("Deny", "")):
313 count = cast(
314 CursorResult,
315 session.execute(
316 delete(ACLPerSource).filter_by(
317 acl=acl, fingerprint=fpr, source=source
318 )
319 ),
320 ).rowcount
321 if count == 0: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 raise CommandError(
323 "Tried to remove upload permissions for package {0}, "
324 "but no upload permissions were granted before.".format(source)
325 )
327 self.log.log(["dm", "deny", fpr.fingerprint, source])
328 self.result.append("Denied: {0}".format(source))
330 session.commit()
332 def _action_dm_admin_common(
333 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
334 ) -> None:
335 cnf = Config()
337 if (
338 "Command::DM-Admin::AdminFingerprints" not in cnf
339 or "Command::DM::ACL" not in cnf
340 ):
341 raise CommandError("DM admin command is not configured for this archive.")
343 allowed_fingerprints = cnf.value_list("Command::DM-Admin::AdminFingerprints")
344 if fingerprint.fingerprint not in allowed_fingerprints:
345 raise CommandError(
346 "Key {0} is not allowed to admin DM".format(fingerprint.fingerprint)
347 )
349 def action_dm_remove(
350 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
351 ) -> None:
352 self._action_dm_admin_common(fingerprint, section, session)
354 cnf = Config()
355 acl_name = cnf.get("Command::DM::ACL", "dm")
356 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one()
358 fpr_hash = section["Fingerprint"].replace(" ", "")
359 fpr = session.scalars(
360 select(Fingerprint).filter_by(fingerprint=fpr_hash).limit(1)
361 ).first()
362 if fpr is None:
363 self.result.append(
364 "Unknown fingerprint: {0}\nNo action taken.".format(fpr_hash)
365 )
366 return
368 self.log.log(["dm-remove", fpr.fingerprint])
370 count = 0
371 for entry in session.scalars(
372 select(ACLPerSource).filter_by(acl=acl, fingerprint=fpr)
373 ):
374 self.log.log(
375 ["dm-remove", fpr.fingerprint, "source={0}".format(entry.source)]
376 )
377 count += 1
378 session.delete(entry)
380 self.result.append(
381 "Removed: {0}.\n{1} acl entries removed.".format(fpr.fingerprint, count)
382 )
384 session.commit()
386 def action_dm_migrate(
387 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
388 ) -> None:
389 self._action_dm_admin_common(fingerprint, section, session)
390 cnf = Config()
391 acl_name = cnf.get("Command::DM::ACL", "dm")
392 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one()
394 fpr_hash_from = section["From"].replace(" ", "")
395 fpr_from = session.scalars(
396 select(Fingerprint).filter_by(fingerprint=fpr_hash_from).limit(1)
397 ).first()
398 if fpr_from is None:
399 self.result.append(
400 "Unknown fingerprint (From): {0}\nNo action taken.".format(
401 fpr_hash_from
402 )
403 )
404 return
406 fpr_hash_to = section["To"].replace(" ", "")
407 fpr_to = session.scalars(
408 select(Fingerprint).filter_by(fingerprint=fpr_hash_to).limit(1)
409 ).first()
410 if fpr_to is None:
411 self.result.append(
412 "Unknown fingerprint (To): {0}\nNo action taken.".format(fpr_hash_to)
413 )
414 return
415 if fpr_to.keyring is None or fpr_to.keyring.keyring_name not in cnf.value_list(
416 "Command::DM::Keyrings"
417 ):
418 self.result.append(
419 "Key (To) {0} is not in DM keyring.\nNo action taken.".format(
420 fpr_to.fingerprint
421 )
422 )
423 return
425 self.log.log(
426 [
427 "dm-migrate",
428 "from={0}".format(fpr_hash_from),
429 "to={0}".format(fpr_hash_to),
430 ]
431 )
433 sources = []
434 for entry in session.scalars(
435 select(ACLPerSource).filter_by(acl=acl, fingerprint=fpr_from)
436 ):
437 self.log.log(
438 [
439 "dm-migrate",
440 "from={0}".format(fpr_hash_from),
441 "to={0}".format(fpr_hash_to),
442 "source={0}".format(entry.source),
443 ]
444 )
445 entry.fingerprint = fpr_to
446 sources.append(entry.source)
448 self.result.append(
449 "Migrated {0} to {1}.\n{2} acl entries changed: {3}".format(
450 fpr_hash_from, fpr_hash_to, len(sources), ", ".join(sources)
451 )
452 )
454 session.commit()
456 def action_break_the_archive(
457 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
458 ) -> None:
459 name = "Dave"
460 uid = fingerprint.uid
461 if uid is not None and uid.name is not None:
462 name = uid.name.split()[0]
464 self.result.append(
465 "DAK9000: I'm sorry, {0}. I'm afraid I can't do that.".format(name)
466 )
468 def _sourcename_from_dbchanges(self, changes: DBChange) -> str:
469 source = changes.source
470 # in case the Source contains spaces, e.g. in binNMU .changes
471 source = source.split(" ")[0]
473 return source
475 def _process_upload_add_command_file(
476 self, upload: PolicyQueueUpload, command: str
477 ) -> None:
478 source = self._sourcename_from_dbchanges(upload.changes)
479 filename = f"{command}.{source}_{upload.changes.version}"
480 content = "OK" if command == "ACCEPT" else "NOTOK"
482 with open(
483 os.path.join(upload.policy_queue.path, "COMMENTS", filename), "x"
484 ) as f:
485 f.write(content + "\n")
487 def _action_process_upload_common(
488 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
489 ) -> None:
490 cnf = Config()
492 if "Command::ProcessUpload::ACL" not in cnf: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true
493 raise CommandError(
494 "Process Upload command is not configured for this archive."
495 )
497 def action_process_upload(
498 self, fingerprint: Fingerprint, section: apt_pkg.TagSection, session: "Session"
499 ) -> None:
500 self._action_process_upload_common(fingerprint, section, session)
502 cnf = Config()
503 acl_name = cnf.get("Command::ProcessUpload::ACL", "process-upload")
504 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one()
506 source = section["Source"].replace(" ", "")
507 version = section["Version"].replace(" ", "")
508 command = section["Command"].replace(" ", "")
510 if command not in ("ACCEPT", "REJECT"): 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true
511 raise CommandError("Invalid ProcessUpload command: {0}".format(command))
513 uploads = session.scalars(
514 select(PolicyQueueUpload)
515 .join(PolicyQueueUpload.changes)
516 .where(DBChange.version == version)
517 ).all()
518 # we don't filter_by(source=source) because a source in a DBChange can
519 # contain more than the source, e.g. 'source (version)' for binNMUs
520 uploads = [
521 upload
522 for upload in uploads
523 if self._sourcename_from_dbchanges(upload.changes) == source
524 ]
525 if not uploads: 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true
526 raise CommandError(
527 "Could not find upload for {0} {1}".format(source, version)
528 )
530 upload = uploads[0]
532 # we consider all uploads except those for NEW, and take into account the
533 # target suite when checking for permissions
534 if upload.policy_queue.queue_name == "new": 534 ↛ 535line 534 didn't jump to line 535 because the condition on line 534 was never true
535 raise CommandError(
536 "Processing uploads from NEW not allowed ({0} {1})".format(
537 source, version
538 )
539 )
541 suite = upload.target_suite
543 self.log.log(
544 [
545 "process-upload",
546 fingerprint.fingerprint,
547 source,
548 version,
549 upload.policy_queue.queue_name,
550 suite.suite_name,
551 ]
552 )
554 per_source_query = select(ACLPerSource).filter_by(
555 acl=acl, fingerprint=fingerprint, source=source
556 )
557 per_suite_query = select(ACLPerSuite).filter_by(
558 acl=acl, fingerprint=fingerprint, suite=suite
559 )
560 allowed = bool(session.scalar(select(per_source_query.exists()))) or bool(
561 session.scalar(select(per_suite_query.exists()))
562 )
564 self.log.log(
565 [
566 "process-upload",
567 fingerprint.fingerprint,
568 source,
569 version,
570 upload.policy_queue.queue_name,
571 suite.suite_name,
572 allowed,
573 ]
574 )
576 if allowed:
577 self._process_upload_add_command_file(upload, command)
579 self.result.append(
580 "ProcessUpload: processed fp {0}: {1}_{2}/{3}".format(
581 fingerprint.fingerprint, source, version, suite.codename
582 )
583 )