Coverage for daklib/command.py: 65%

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

26import apt_pkg

28from daklib.config import Config

29from daklib.dak_exceptions import ParseMaintError

30from daklib.dbconn import (

31 ACL,

32 ACLPerSource,

33 ACLPerSuite,

34 DBChange,

35 DBConn,

36 DBSource,

37 Fingerprint,

38 Keyring,

39 PolicyQueueUpload,

40 SignatureHistory,

41)

42from daklib.gpg import SignedFile

43from daklib.regexes import re_field_package

44from daklib.textutils import fix_maintainer

45from daklib.utils import TemplateSubst, gpg_get_key_addresses, send_mail

48class CommandError(Exception):

49 pass

52class CommandFile:

53 def __init__(self, filename: str, data: bytes, log=None):

54 if log is None: 54 ↛ 55line 54 didn't jump to line 55, because the condition on line 54 was never true

55 from daklib.daklog import Logger

57 log = Logger()

58 self.cc: list[str] = []

59 self.result = []

60 self.log = log

61 self.filename: str = filename

62 self.data = data

63 self.uploader = None

65 def _check_replay(self, signed_file: SignedFile, session):

66 """check for replays

68 .. note::

70 Will commit changes to the database.

72 :param session: database session

73 """

74 # Mark commands file as seen to prevent replays.

75 signature_history = SignatureHistory.from_signed_file(signed_file)

76 session.add(signature_history)

77 session.commit()

79 def _quote_section(self, section) -> str:

80 lines = [f"> {line}" for line in str(section).splitlines()]

81 return "\n".join(lines)

83 def _evaluate_sections(self, sections, session):

84 session.rollback()

85 try:

86 while True:

87 next(sections)

88 section = sections.section

89 self.result.append(self._quote_section(section))

91 action = section.get("Action", None)

92 if action is None: 92 ↛ 93line 92 didn't jump to line 93, because the condition on line 92 was never true

93 raise CommandError("Encountered section without Action field")

95 if action == "dm":

96 self.action_dm(self.fingerprint, section, session)

97 elif action == "dm-remove": 97 ↛ 98line 97 didn't jump to line 98, because the condition on line 97 was never true

98 self.action_dm_remove(self.fingerprint, section, session)

99 elif action == "dm-migrate": 99 ↛ 100line 99 didn't jump to line 100, because the condition on line 99 was never true

100 self.action_dm_migrate(self.fingerprint, section, session)

101 elif action == "break-the-archive": 101 ↛ 102line 101 didn't jump to line 102, because the condition on line 101 was never true

102 self.action_break_the_archive(self.fingerprint, section, session)

103 elif action == "process-upload": 103 ↛ 106line 103 didn't jump to line 106, because the condition on line 103 was never false

104 self.action_process_upload(self.fingerprint, section, session)

105 else:

106 raise CommandError("Unknown action: {0}".format(action))

107

108 self.result.append("")

109 except StopIteration:

110 pass

111 finally:

112 session.rollback()

113

114 def _notify_uploader(self):

115 cnf = Config()

116

117 bcc = "X-DAK: dak process-command"

118 if "Dinstall::Bcc" in cnf: 118 ↛ 119line 118 didn't jump to line 119, because the condition on line 118 was never true

119 bcc = "{0}\nBcc: {1}".format(bcc, cnf["Dinstall::Bcc"])

120

121 maint_to = None

122 addresses = gpg_get_key_addresses(self.fingerprint.fingerprint)

123 if len(addresses) > 0: 123 ↛ 126line 123 didn't jump to line 126, because the condition on line 123 was never false

124 maint_to = addresses[0]

125

126 if self.uploader:

127 try:

128 maint_to = fix_maintainer(self.uploader)[1]

129 except ParseMaintError:

130 self.log.log("ignoring malformed uploader", self.filename)

131

132 cc = set()

133 for address in self.cc:

134 try:

135 cc.add(fix_maintainer(address)[1])

136 except ParseMaintError:

137 self.log.log("ignoring malformed cc", self.filename)

138

139 subst = {

140 "__DAK_ADDRESS__": cnf["Dinstall::MyEmailAddress"],

141 "__MAINTAINER_TO__": maint_to,

142 "__CC__": ", ".join(cc),

143 "__BCC__": bcc,

144 "__RESULTS__": "\n".join(self.result),

145 "__FILENAME__": self.filename,

146 }

147

148 message = TemplateSubst(

149 subst, os.path.join(cnf["Dir::Templates"], "process-command.processed")

150 )

151

152 send_mail(message)

153

154 def evaluate(self) -> bool:

155 """evaluate commands file

156

157 :return: :const:`True` if the file was processed sucessfully,

158 :const:`False` otherwise

159 """

160 result = True

161

162 session = DBConn().session()

163

164 keyrings = (

165 session.query(Keyring).filter_by(active=True).order_by(Keyring.priority)

166 )

167 keyring_files = [k.keyring_name for k in keyrings]

168

169 signed_file = SignedFile(self.data, keyring_files)

170 if not signed_file.valid: 170 ↛ 171line 170 didn't jump to line 171, because the condition on line 170 was never true

171 self.log.log(["invalid signature", self.filename])

172 return False

173

174 self.fingerprint = (

175 session.query(Fingerprint)

176 .filter_by(fingerprint=signed_file.primary_fingerprint)

177 .one()

178 )

179 if self.fingerprint.keyring is None: 179 ↛ 180line 179 didn't jump to line 180, because the condition on line 179 was never true

180 self.log.log(["signed by key in unknown keyring", self.filename])

181 return False

182 assert self.fingerprint.keyring.active

183

184 self.log.log(

185 [

186 "processing",

187 self.filename,

188 "signed-by={0}".format(self.fingerprint.fingerprint),

189 ]

190 )

191

192 with tempfile.TemporaryFile() as fh:

193 fh.write(signed_file.contents)

194 fh.seek(0)

195 sections = apt_pkg.TagFile(fh)

196

197 try:

198 next(sections)

199 section = sections.section

200 if "Uploader" in section:

201 self.uploader = section["Uploader"]

202 if "Cc" in section: 202 ↛ 203line 202 didn't jump to line 203, because the condition on line 202 was never true

203 self.cc.append(section["Cc"])

204 # TODO: Verify first section has valid Archive field

205 if "Archive" not in section: 205 ↛ 206line 205 didn't jump to line 206, because the condition on line 205 was never true

206 raise CommandError("No Archive field in first section.")

207

208 # TODO: send mail when we detected a replay.

209 self._check_replay(signed_file, session)

210

211 self._evaluate_sections(sections, session)

212 self.result.append("")

213 except Exception as e:

214 self.log.log(["ERROR", e])

215 self.result.append(

216 "There was an error processing this section. No changes were committed.\nDetails:\n{0}".format(

217 e

218 )

219 )

220 result = False

221

222 self._notify_uploader()

223

224 session.close()

225

226 return result

227

228 def _split_packages(self, value: str) -> list[str]:

229 names = value.split()

230 for name in names:

231 if not re_field_package.match(name): 231 ↛ 232line 231 didn't jump to line 232, because the condition on line 231 was never true

232 raise CommandError('Invalid package name "{0}"'.format(name))

233 return names

234

235 def action_dm(self, fingerprint, section, session) -> None:

236 cnf = Config()

237

238 if ( 238 ↛ 243line 238 didn't jump to line 243

239 "Command::DM::AdminKeyrings" not in cnf

240 or "Command::DM::ACL" not in cnf

241 or "Command::DM::Keyrings" not in cnf

242 ):

243 raise CommandError("DM command is not configured for this archive.")

244

245 allowed_keyrings = cnf.value_list("Command::DM::AdminKeyrings")

246 if fingerprint.keyring.keyring_name not in allowed_keyrings: 246 ↛ 247line 246 didn't jump to line 247, because the condition on line 246 was never true

247 raise CommandError(

248 "Key {0} is not allowed to set DM".format(fingerprint.fingerprint)

249 )

250

251 acl_name = cnf.get("Command::DM::ACL", "dm")

252 acl = session.query(ACL).filter_by(name=acl_name).one()

253

254 fpr_hash = section["Fingerprint"].replace(" ", "")

255 fpr = session.query(Fingerprint).filter_by(fingerprint=fpr_hash).first()

256 if fpr is None: 256 ↛ 257line 256 didn't jump to line 257, because the condition on line 256 was never true

257 raise CommandError("Unknown fingerprint {0}".format(fpr_hash))

258 if fpr.keyring is None or fpr.keyring.keyring_name not in cnf.value_list(

259 "Command::DM::Keyrings"

260 ):

261 raise CommandError("Key {0} is not in DM keyring.".format(fpr.fingerprint))

262 addresses = gpg_get_key_addresses(fpr.fingerprint)

263 if len(addresses) > 0: 263 ↛ 266line 263 didn't jump to line 266, because the condition on line 263 was never false

264 self.cc.append(addresses[0])

265

266 self.log.log(["dm", "fingerprint", fpr.fingerprint])

267 self.result.append("Fingerprint: {0}".format(fpr.fingerprint))

268 if len(addresses) > 0: 268 ↛ 272line 268 didn't jump to line 272, because the condition on line 268 was never false

269 self.log.log(["dm", "uid", addresses[0]])

270 self.result.append("Uid: {0}".format(addresses[0]))

271

272 for source in self._split_packages(section.get("Allow", "")):

273 # Check for existance of source package to catch typos

274 if session.query(DBSource).filter_by(source=source).first() is None: 274 ↛ 275line 274 didn't jump to line 275, because the condition on line 274 was never true