Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/scapy/layers/tftp.py: 44%

1# SPDX-License-Identifier: GPL-2.0-only 

2# This file is part of Scapy 

3# See https://scapy.net/ for more information 

4# Copyright (C) Philippe Biondi <phil@secdev.org> 

5 

6""" 

7TFTP (Trivial File Transfer Protocol). 

8""" 

9 

10import os 

11import random 

12 

13from scapy.packet import Packet, bind_layers, split_bottom_up, bind_bottom_up 

14from scapy.fields import PacketListField, ShortEnumField, ShortField, \ 

15 StrNullField 

16from scapy.automaton import ATMT, Automaton 

17from scapy.layers.inet import UDP, IP 

18from scapy.config import conf 

19from scapy.volatile import RandShort 

20 

21 

22TFTP_operations = {1: "RRQ", 2: "WRQ", 3: "DATA", 4: "ACK", 5: "ERROR", 6: "OACK"} # noqa: E501 

23 

24 

25class TFTP(Packet): 

26 name = "TFTP opcode" 

27 fields_desc = [ShortEnumField("op", 1, TFTP_operations), ] 

28 

29 

30class TFTP_RRQ(Packet): 

31 name = "TFTP Read Request" 

32 fields_desc = [StrNullField("filename", ""), 

33 StrNullField("mode", "octet")] 

34 

35 def answers(self, other): 

36 return 0 

37 

38 def mysummary(self): 

39 return self.sprintf("RRQ %filename%"), [UDP] 

40 

41 

42class TFTP_WRQ(Packet): 

43 name = "TFTP Write Request" 

44 fields_desc = [StrNullField("filename", ""), 

45 StrNullField("mode", "octet")] 

46 

47 def answers(self, other): 

48 return 0 

49 

50 def mysummary(self): 

51 return self.sprintf("WRQ %filename%"), [UDP] 

52 

53 

54class TFTP_DATA(Packet): 

55 name = "TFTP Data" 

56 fields_desc = [ShortField("block", 0)] 

57 

58 def answers(self, other): 

59 return self.block == 1 and isinstance(other, TFTP_RRQ) 

60 

61 def mysummary(self): 

62 return self.sprintf("DATA %block%"), [UDP] 

63 

64 

65class TFTP_Option(Packet): 

66 fields_desc = [StrNullField("oname", ""), 

67 StrNullField("value", "")] 

68 

69 def extract_padding(self, pkt): 

70 return "", pkt 

71 

72 

73class TFTP_Options(Packet): 

74 fields_desc = [PacketListField("options", [], TFTP_Option, length_from=lambda x:None)] # noqa: E501 

75 

76 

77class TFTP_ACK(Packet): 

78 name = "TFTP Ack" 

79 fields_desc = [ShortField("block", 0)] 

80 

81 def answers(self, other): 

82 if isinstance(other, TFTP_DATA): 

83 return self.block == other.block 

84 elif isinstance(other, (TFTP_RRQ, TFTP_WRQ, TFTP_OACK)): # noqa: E501 

85 return self.block == 0 

86 return 0 

87 

88 def mysummary(self): 

89 return self.sprintf("ACK %block%"), [UDP] 

90 

91 

92TFTP_Error_Codes = {0: "Not defined", 

93 1: "File not found", 

94 2: "Access violation", 

95 3: "Disk full or allocation exceeded", 

96 4: "Illegal TFTP operation", 

97 5: "Unknown transfer ID", 

98 6: "File already exists", 

99 7: "No such user", 

100 8: "Terminate transfer due to option negotiation", 

101 } 

102 

103 

104class TFTP_ERROR(Packet): 

105 name = "TFTP Error" 

106 fields_desc = [ShortEnumField("errorcode", 0, TFTP_Error_Codes), 

107 StrNullField("errormsg", "")] 

108 

109 def answers(self, other): 

110 return isinstance(other, (TFTP_DATA, TFTP_RRQ, TFTP_WRQ, TFTP_ACK)) 

111 

112 def mysummary(self): 

113 return self.sprintf("ERROR %errorcode%: %errormsg%"), [UDP] 

114 

115 

116class TFTP_OACK(Packet): 

117 name = "TFTP Option Ack" 

118 fields_desc = [] 

119 

120 def answers(self, other): 

121 return isinstance(other, (TFTP_WRQ, TFTP_RRQ)) 

122 

123 

124bind_layers(UDP, TFTP, dport=69) 

125bind_layers(TFTP, TFTP_RRQ, op=1) 

126bind_layers(TFTP, TFTP_WRQ, op=2) 

127bind_layers(TFTP, TFTP_DATA, op=3) 

128bind_layers(TFTP, TFTP_ACK, op=4) 

129bind_layers(TFTP, TFTP_ERROR, op=5) 

130bind_layers(TFTP, TFTP_OACK, op=6) 

131bind_layers(TFTP_RRQ, TFTP_Options) 

132bind_layers(TFTP_WRQ, TFTP_Options) 

133bind_layers(TFTP_OACK, TFTP_Options) 

134 

135 

136class TFTP_read(Automaton): 

137 def parse_args(self, filename, server, sport=None, port=69, **kargs): 

138 Automaton.parse_args(self, **kargs) 

139 self.filename = filename 

140 self.server = server 

141 self.port = port 

142 self.sport = sport 

143 

144 def master_filter(self, pkt): 

145 return (IP in pkt and pkt[IP].src == self.server and UDP in pkt and 

146 pkt[UDP].dport == self.my_tid and 

147 (self.server_tid is None or pkt[UDP].sport == self.server_tid)) 

148 

149 # BEGIN 

150 @ATMT.state(initial=1) 

151 def BEGIN(self): 

152 self.blocksize = 512 

153 self.my_tid = self.sport or RandShort()._fix() 

154 bind_bottom_up(UDP, TFTP, dport=self.my_tid) 

155 self.server_tid = None 

156 self.res = b"" 

157 

158 self.l3 = IP(dst=self.server) / UDP(sport=self.my_tid, dport=self.port) / TFTP() # noqa: E501 

159 self.last_packet = self.l3 / TFTP_RRQ(filename=self.filename, mode="octet") # noqa: E501 

160 self.send(self.last_packet) 

161 self.awaiting = 1 

162 

163 raise self.WAITING() 

164 

165 # WAITING 

166 @ATMT.state() 

167 def WAITING(self): 

168 pass 

169 

170 @ATMT.receive_condition(WAITING) 

171 def receive_data(self, pkt): 

172 if TFTP_DATA in pkt and pkt[TFTP_DATA].block == self.awaiting: 

173 if self.server_tid is None: 

174 self.server_tid = pkt[UDP].sport 

175 self.l3[UDP].dport = self.server_tid 

176 raise self.RECEIVING(pkt) 

177 

178 @ATMT.receive_condition(WAITING, prio=1) 

179 def receive_error(self, pkt): 

180 if TFTP_ERROR in pkt: 

181 raise self.ERROR(pkt) 

182 

183 @ATMT.timeout(WAITING, 3) 

184 def timeout_waiting(self): 

185 raise self.WAITING() 

186 

187 @ATMT.action(timeout_waiting) 

188 def retransmit_last_packet(self): 

189 self.send(self.last_packet) 

190 

191 @ATMT.action(receive_data) 

192# @ATMT.action(receive_error) 

193 def send_ack(self): 

194 self.last_packet = self.l3 / TFTP_ACK(block=self.awaiting) 

195 self.send(self.last_packet) 

196 

197 # RECEIVED 

198 @ATMT.state() 

199 def RECEIVING(self, pkt): 

200 if conf.raw_layer in pkt: 

201 recvd = pkt[conf.raw_layer].load 

202 else: 

203 recvd = b"" 

204 self.res += recvd 

205 self.awaiting += 1 

206 if len(recvd) == self.blocksize: 

207 raise self.WAITING() 

208 raise self.END() 

209 

210 # ERROR 

211 @ATMT.state(error=1) 

212 def ERROR(self, pkt): 

213 split_bottom_up(UDP, TFTP, dport=self.my_tid) 

214 return pkt[TFTP_ERROR].summary() 

215 

216 # END 

217 @ATMT.state(final=1) 

218 def END(self): 

219 split_bottom_up(UDP, TFTP, dport=self.my_tid) 

220 return self.res 

221 

222 

223class TFTP_write(Automaton): 

224 def parse_args(self, filename, data, server, sport=None, port=69, **kargs): 

225 Automaton.parse_args(self, **kargs) 

226 self.filename = filename 

227 self.server = server 

228 self.port = port 

229 self.sport = sport 

230 self.blocksize = 512 

231 self.origdata = data 

232 

233 def master_filter(self, pkt): 

234 return (IP in pkt and pkt[IP].src == self.server and UDP in pkt and 

235 pkt[UDP].dport == self.my_tid and 

236 (self.server_tid is None or pkt[UDP].sport == self.server_tid)) 

237 

238 # BEGIN 

239 @ATMT.state(initial=1) 

240 def BEGIN(self): 

241 self.data = [self.origdata[i * self.blocksize:(i + 1) * self.blocksize] 

242 for i in range(len(self.origdata) // self.blocksize + 1)] 

243 self.my_tid = self.sport or RandShort()._fix() 

244 bind_bottom_up(UDP, TFTP, dport=self.my_tid) 

245 self.server_tid = None 

246 

247 self.l3 = IP(dst=self.server) / UDP(sport=self.my_tid, dport=self.port) / TFTP() # noqa: E501 

248 self.last_packet = self.l3 / TFTP_WRQ(filename=self.filename, mode="octet") # noqa: E501 

249 self.send(self.last_packet) 

250 self.res = "" 

251 self.awaiting = 0 

252 

253 raise self.WAITING_ACK() 

254 

255 # WAITING_ACK 

256 @ATMT.state() 

257 def WAITING_ACK(self): 

258 pass 

259 

260 @ATMT.receive_condition(WAITING_ACK) 

261 def received_ack(self, pkt): 

262 if TFTP_ACK in pkt and pkt[TFTP_ACK].block == self.awaiting: 

263 if self.server_tid is None: 

264 self.server_tid = pkt[UDP].sport 

265 self.l3[UDP].dport = self.server_tid 

266 raise self.SEND_DATA() 

267 

268 @ATMT.receive_condition(WAITING_ACK) 

269 def received_error(self, pkt): 

270 if TFTP_ERROR in pkt: 

271 raise self.ERROR(pkt) 

272 

273 @ATMT.timeout(WAITING_ACK, 3) 

274 def timeout_waiting(self): 

275 raise self.WAITING_ACK() 

276 

277 @ATMT.action(timeout_waiting) 

278 def retransmit_last_packet(self): 

279 self.send(self.last_packet) 

280 

281 # SEND_DATA 

282 @ATMT.state() 

283 def SEND_DATA(self): 

284 self.awaiting += 1 

285 self.last_packet = self.l3 / TFTP_DATA(block=self.awaiting) / self.data.pop(0) # noqa: E501 

286 self.send(self.last_packet) 

287 if self.data: 

288 raise self.WAITING_ACK() 

289 raise self.END() 

290 

291 # ERROR 

292 @ATMT.state(error=1) 

293 def ERROR(self, pkt): 

294 split_bottom_up(UDP, TFTP, dport=self.my_tid) 

295 return pkt[TFTP_ERROR].summary() 

296 

297 # END 

298 @ATMT.state(final=1) 

299 def END(self): 

300 split_bottom_up(UDP, TFTP, dport=self.my_tid) 

301 

302 

303class TFTP_WRQ_server(Automaton): 

304 

305 def parse_args(self, ip=None, sport=None, *args, **kargs): 

306 Automaton.parse_args(self, *args, **kargs) 

307 self.ip = ip 

308 self.sport = sport 

309 

310 def master_filter(self, pkt): 

311 return TFTP in pkt and (not self.ip or pkt[IP].dst == self.ip) 

312 

313 @ATMT.state(initial=1) 

314 def BEGIN(self): 

315 self.blksize = 512 

316 self.blk = 1 

317 self.filedata = b"" 

318 self.my_tid = self.sport or random.randint(10000, 65500) 

319 bind_bottom_up(UDP, TFTP, dport=self.my_tid) 

320 

321 @ATMT.receive_condition(BEGIN) 

322 def receive_WRQ(self, pkt): 

323 if TFTP_WRQ in pkt: 

324 raise self.WAIT_DATA().action_parameters(pkt) 

325 

326 @ATMT.action(receive_WRQ) 

327 def ack_WRQ(self, pkt): 

328 ip = pkt[IP] 

329 self.ip = ip.dst 

330 self.dst = ip.src 

331 self.filename = pkt[TFTP_WRQ].filename 

332 options = pkt.getlayer(TFTP_Options) 

333 self.l3 = IP(src=ip.dst, dst=ip.src) / UDP(sport=self.my_tid, dport=pkt.sport) / TFTP() # noqa: E501 

334 if options is None: 

335 self.last_packet = self.l3 / TFTP_ACK(block=0) 

336 self.send(self.last_packet) 

337 else: 

338 opt = [x for x in options.options if x.oname.upper() == b"BLKSIZE"] 

339 if opt: 

340 self.blksize = int(opt[0].value) 

341 self.debug(2, "Negotiated new blksize at %i" % self.blksize) 

342 self.last_packet = self.l3 / TFTP_OACK() / TFTP_Options(options=opt) # noqa: E501 

343 self.send(self.last_packet) 

344 

345 @ATMT.state() 

346 def WAIT_DATA(self): 

347 pass 

348 

349 @ATMT.timeout(WAIT_DATA, 1) 

350 def resend_ack(self): 

351 self.send(self.last_packet) 

352 raise self.WAIT_DATA() 

353 

354 @ATMT.receive_condition(WAIT_DATA) 

355 def receive_data(self, pkt): 

356 if TFTP_DATA in pkt: 

357 data = pkt[TFTP_DATA] 

358 if data.block == self.blk: 

359 raise self.DATA(data) 

360 

361 @ATMT.action(receive_data) 

362 def ack_data(self): 

363 self.last_packet = self.l3 / TFTP_ACK(block=self.blk) 

364 self.send(self.last_packet) 

365 

366 @ATMT.state() 

367 def DATA(self, data): 

368 self.filedata += data.load 

369 if len(data.load) < self.blksize: 

370 raise self.END() 

371 self.blk += 1 

372 raise self.WAIT_DATA() 

373 

374 @ATMT.state(final=1) 

375 def END(self): 

376 split_bottom_up(UDP, TFTP, dport=self.my_tid) 

377 return self.filename, self.filedata 

378 

379 

380class TFTP_RRQ_server(Automaton): 

381 def parse_args(self, store=None, joker=None, dir=None, ip=None, sport=None, serve_one=False, **kargs): # noqa: E501 

382 Automaton.parse_args(self, **kargs) 

383 if store is None: 

384 store = {} 

385 if dir is not None: 

386 self.dir = os.path.join(os.path.abspath(dir), "") 

387 else: 

388 self.dir = None 

389 self.store = store 

390 self.joker = joker 

391 self.ip = ip 

392 self.sport = sport 

393 self.serve_one = serve_one 

394 self.my_tid = self.sport or random.randint(10000, 65500) 

395 bind_bottom_up(UDP, TFTP, dport=self.my_tid) 

396 

397 def master_filter(self, pkt): 

398 return TFTP in pkt and (not self.ip or pkt[IP].dst == self.ip) 

399 

400 @ATMT.state(initial=1) 

401 def WAIT_RRQ(self): 

402 self.blksize = 512 

403 self.blk = 0 

404 

405 @ATMT.receive_condition(WAIT_RRQ) 

406 def receive_rrq(self, pkt): 

407 if TFTP_RRQ in pkt: 

408 raise self.RECEIVED_RRQ(pkt) 

409 

410 @ATMT.state() 

411 def RECEIVED_RRQ(self, pkt): 

412 ip = pkt[IP] 

413 options = pkt[TFTP_Options] 

414 self.l3 = IP(src=ip.dst, dst=ip.src) / UDP(sport=self.my_tid, dport=ip.sport) / TFTP() # noqa: E501 

415 self.filename = pkt[TFTP_RRQ].filename.decode("utf-8", "ignore") 

416 self.blk = 1 

417 self.data = None 

418 if self.filename in self.store: 

419 self.data = self.store[self.filename] 

420 elif self.dir is not None: 

421 fn = os.path.abspath(os.path.join(self.dir, self.filename)) 

422 if fn.startswith(self.dir): # Check we're still in the server's directory # noqa: E501 

423 try: 

424 with open(fn) as fd: 

425 self.data = fd.read() 

426 except IOError: 

427 pass 

428 if self.data is None: 

429 self.data = self.joker 

430 

431 if options: 

432 opt = [x for x in options.options if x.oname.upper() == b"BLKSIZE"] 

433 if opt: 

434 self.blksize = int(opt[0].value) 

435 self.debug(2, "Negotiated new blksize at %i" % self.blksize) 

436 self.last_packet = self.l3 / TFTP_OACK() / TFTP_Options(options=opt) # noqa: E501 

437 self.send(self.last_packet) 

438 

439 @ATMT.condition(RECEIVED_RRQ) 

440 def file_in_store(self): 

441 if self.data is not None: 

442 self.blknb = len(self.data) / self.blksize + 1 

443 raise self.SEND_FILE() 

444 

445 @ATMT.condition(RECEIVED_RRQ) 

446 def file_not_found(self): 

447 if self.data is None: 

448 raise self.WAIT_RRQ() 

449 

450 @ATMT.action(file_not_found) 

451 def send_error(self): 

452 self.send(self.l3 / TFTP_ERROR(errorcode=1, errormsg=TFTP_Error_Codes[1])) # noqa: E501 

453 

454 @ATMT.state() 

455 def SEND_FILE(self): 

456 self.send(self.l3 / TFTP_DATA(block=self.blk) / self.data[(self.blk - 1) * self.blksize:self.blk * self.blksize]) # noqa: E501 

457 

458 @ATMT.timeout(SEND_FILE, 3) 

459 def timeout_waiting_ack(self): 

460 raise self.SEND_FILE() 

461 

462 @ATMT.receive_condition(SEND_FILE) 

463 def received_ack(self, pkt): 

464 if TFTP_ACK in pkt and pkt[TFTP_ACK].block == self.blk: 

465 raise self.RECEIVED_ACK() 

466 

467 @ATMT.state() 

468 def RECEIVED_ACK(self): 

469 self.blk += 1 

470 

471 @ATMT.condition(RECEIVED_ACK) 

472 def no_more_data(self): 

473 if self.blk > self.blknb: 

474 if self.serve_one: 

475 raise self.END() 

476 raise self.WAIT_RRQ() 

477 

478 @ATMT.condition(RECEIVED_ACK, prio=2) 

479 def data_remaining(self): 

480 raise self.SEND_FILE() 

481 

482 @ATMT.state(final=1) 

483 def END(self): 

484 split_bottom_up(UDP, TFTP, dport=self.my_tid)