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

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 

9This provides TFTP implementation and 4 small automata: 

10 - TFTP_read: read a remote file 

11 - TFTP_RRQ_server: server that answers to read requests 

12 - TFTP_write: write a remote file 

13 - TFTP_WRQ_server: server than accepts write requests 

14""" 

15 

16import os 

17import random 

18 

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

20from scapy.fields import ( 

21 PacketListField, 

22 ShortEnumField, 

23 ShortField, 

24 StrNullField, 

25) 

26from scapy.automaton import ATMT, Automaton 

27from scapy.base_classes import Net 

28from scapy.config import conf 

29from scapy.sessions import IPSession 

30from scapy.volatile import RandShort 

31 

32from scapy.layers.inet import UDP, IP 

33 

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

35 

36 

37class TFTP(Packet): 

38 name = "TFTP opcode" 

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

40 

41 

42class TFTP_RRQ(Packet): 

43 name = "TFTP Read 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("RRQ %filename%"), [UDP] 

52 

53 

54class TFTP_WRQ(Packet): 

55 name = "TFTP Write Request" 

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

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

58 

59 def answers(self, other): 

60 return 0 

61 

62 def mysummary(self): 

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

64 

65 

66class TFTP_DATA(Packet): 

67 name = "TFTP Data" 

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

69 

70 def answers(self, other): 

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

72 

73 def mysummary(self): 

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

75 

76 

77class TFTP_Option(Packet): 

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

79 StrNullField("value", "")] 

80 

81 def extract_padding(self, pkt): 

82 return "", pkt 

83 

84 

85class TFTP_Options(Packet): 

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

87 

88 

89class TFTP_ACK(Packet): 

90 name = "TFTP Ack" 

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

92 

93 def answers(self, other): 

94 if isinstance(other, TFTP_DATA): 

95 return self.block == other.block 

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

97 return self.block == 0 

98 return 0 

99 

100 def mysummary(self): 

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

102 

103 

104TFTP_Error_Codes = {0: "Not defined", 

105 1: "File not found", 

106 2: "Access violation", 

107 3: "Disk full or allocation exceeded", 

108 4: "Illegal TFTP operation", 

109 5: "Unknown transfer ID", 

110 6: "File already exists", 

111 7: "No such user", 

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

113 } 

114 

115 

116class TFTP_ERROR(Packet): 

117 name = "TFTP Error" 

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

119 StrNullField("errormsg", "")] 

120 

121 def answers(self, other): 

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

123 

124 def mysummary(self): 

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

126 

127 

128class TFTP_OACK(Packet): 

129 name = "TFTP Option Ack" 

130 fields_desc = [] 

131 

132 def answers(self, other): 

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

134 

135 

136bind_layers(UDP, TFTP, dport=69) 

137bind_layers(TFTP, TFTP_RRQ, op=1) 

138bind_layers(TFTP, TFTP_WRQ, op=2) 

139bind_layers(TFTP, TFTP_DATA, op=3) 

140bind_layers(TFTP, TFTP_ACK, op=4) 

141bind_layers(TFTP, TFTP_ERROR, op=5) 

142bind_layers(TFTP, TFTP_OACK, op=6) 

143bind_layers(TFTP_RRQ, TFTP_Options) 

144bind_layers(TFTP_WRQ, TFTP_Options) 

145bind_layers(TFTP_OACK, TFTP_Options) 

146 

147 

148# Automatons 

149 

150class TFTP_read(Automaton): 

151 """ 

152 TFTP automaton to read a remote file on a TFTP server. 

153 

154 :param filename: the name of the remote file to read. 

155 :param server: the host on which to read (IP or name). 

156 :param sport: (optional) the source port to use. (default: random) 

157 :param port: (optional) the TFTP port (default: 69) 

158 """ 

159 

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

161 if "iface" not in kargs: 

162 server = str(Net(server)) 

163 kargs["iface"] = conf.route.route(server)[0] 

164 Automaton.parse_args(self, **kargs) 

165 self.filename = filename 

166 self.server = server 

167 self.port = port 

168 self.sport = sport 

169 

170 def master_filter(self, pkt): 

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

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

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

174 

175 # BEGIN 

176 @ATMT.state(initial=1) 

177 def BEGIN(self): 

178 self.blocksize = 512 

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

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

181 self.server_tid = None 

182 self.res = bytearray() 

183 

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

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

186 self.send(self.last_packet) 

187 self.awaiting = 1 

188 

189 raise self.WAITING() 

190 

191 # WAITING 

192 @ATMT.state() 

193 def WAITING(self): 

194 pass 

195 

196 @ATMT.receive_condition(WAITING) 

197 def receive_data(self, pkt): 

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

199 if self.server_tid is None: 

200 self.server_tid = pkt[UDP].sport 

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

202 raise self.RECEIVING(pkt) 

203 

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

205 def receive_error(self, pkt): 

206 if TFTP_ERROR in pkt: 

207 raise self.ERROR(pkt) 

208 

209 @ATMT.timeout(WAITING, 3) 

210 def timeout_waiting(self): 

211 raise self.WAITING() 

212 

213 @ATMT.action(timeout_waiting) 

214 def retransmit_last_packet(self): 

215 self.send(self.last_packet) 

216 

217 @ATMT.action(receive_data) 

218# @ATMT.action(receive_error) 

219 def send_ack(self): 

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

221 self.send(self.last_packet) 

222 

223 # RECEIVED 

224 @ATMT.state() 

225 def RECEIVING(self, pkt): 

226 if conf.raw_layer in pkt: 

227 recvd = pkt[conf.raw_layer].load 

228 else: 

229 recvd = b"" 

230 self.res += recvd 

231 self.awaiting += 1 

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

233 raise self.WAITING() 

234 raise self.END() 

235 

236 # ERROR 

237 @ATMT.state(error=1) 

238 def ERROR(self, pkt): 

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

240 return pkt[TFTP_ERROR].summary() 

241 

242 # END 

243 @ATMT.state(final=1) 

244 def END(self): 

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

246 return bytes(self.res) 

247 

248 

249class TFTP_write(Automaton): 

250 """ 

251 TFTP automaton to write a local file onto a TFTP server. 

252 

253 :param filename: the name of the remote file to write. 

254 :param data: the bytes data to write. 

255 :param server: the host on which to read (IP or name). 

256 :param sport: (optional) the source port to use. (default: random) 

257 :param port: (optional) the TFTP port (default: 69) 

258 """ 

259 

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

261 if "iface" not in kargs: 

262 server = str(Net(server)) 

263 kargs["iface"] = conf.route.route(server)[0] 

264 Automaton.parse_args(self, **kargs) 

265 self.filename = filename 

266 self.server = server 

267 self.port = port 

268 self.sport = sport 

269 self.blocksize = 512 

270 self.origdata = data 

271 

272 def master_filter(self, pkt): 

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

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

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

276 

277 # BEGIN 

278 @ATMT.state(initial=1) 

279 def BEGIN(self): 

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

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

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

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

284 self.server_tid = None 

285 

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

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

288 self.send(self.last_packet) 

289 self.res = "" 

290 self.awaiting = 0 

291 

292 raise self.WAITING_ACK() 

293 

294 # WAITING_ACK 

295 @ATMT.state() 

296 def WAITING_ACK(self): 

297 pass 

298 

299 @ATMT.receive_condition(WAITING_ACK) 

300 def received_ack(self, pkt): 

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

302 if self.server_tid is None: 

303 self.server_tid = pkt[UDP].sport 

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

305 raise self.SEND_DATA() 

306 

307 @ATMT.receive_condition(WAITING_ACK) 

308 def received_error(self, pkt): 

309 if TFTP_ERROR in pkt: 

310 raise self.ERROR(pkt) 

311 

312 @ATMT.timeout(WAITING_ACK, 3) 

313 def timeout_waiting(self): 

314 raise self.WAITING_ACK() 

315 

316 @ATMT.action(timeout_waiting) 

317 def retransmit_last_packet(self): 

318 self.send(self.last_packet) 

319 

320 # SEND_DATA 

321 @ATMT.state() 

322 def SEND_DATA(self): 

323 self.awaiting += 1 

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

325 self.send(self.last_packet) 

326 if self.data: 

327 raise self.WAITING_ACK() 

328 raise self.END() 

329 

330 # ERROR 

331 @ATMT.state(error=1) 

332 def ERROR(self, pkt): 

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

334 return pkt[TFTP_ERROR].summary() 

335 

336 # END 

337 @ATMT.state(final=1) 

338 def END(self): 

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

340 

341 

342class TFTP_WRQ_server(Automaton): 

343 """ 

344 TFTP automaton to wait for incoming files 

345 

346 :param ip: (optional) the local IP to listen on. 

347 :param sport: (optional) the local port (by default: random) 

348 """ 

349 

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

351 if "iface" not in kargs and ip: 

352 ip = str(Net(ip)) 

353 kargs["iface"] = conf.route.route(ip)[0] 

354 kargs.setdefault("session", IPSession()) 

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

356 self.ip = ip 

357 self.sport = sport 

358 

359 def master_filter(self, pkt): 

360 return ( 

361 TFTP in pkt and 

362 (not self.ip or pkt[IP].dst == self.ip) and 

363 (self.client is None or 

364 self.client == (pkt[IP].src, pkt[UDP].sport)) 

365 ) 

366 

367 @ATMT.state(initial=1) 

368 def BEGIN(self): 

369 self.blksize = 512 

370 self.blk = 1 

371 self.filedata = b"" 

372 self.client = None 

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

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

375 

376 @ATMT.receive_condition(BEGIN) 

377 def receive_WRQ(self, pkt): 

378 if TFTP_WRQ in pkt: 

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

380 

381 @ATMT.action(receive_WRQ) 

382 def ack_WRQ(self, pkt): 

383 ip = pkt[IP] 

384 self.ip = ip.dst 

385 self.dst = ip.src 

386 self.client = (ip.src, pkt[UDP].sport) 

387 self.filename = pkt[TFTP_WRQ].filename 

388 options = pkt.getlayer(TFTP_Options) 

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

390 if options is None: 

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

392 self.send(self.last_packet) 

393 else: 

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

395 if opt: 

396 try: 

397 blksize = int(opt[0].value) 

398 if blksize < 8 or blksize > 65464: 

399 raise ValueError 

400 self.blksize = blksize 

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

402 except ValueError: 

403 opt = [] 

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

405 self.send(self.last_packet) 

406 

407 @ATMT.state() 

408 def WAIT_DATA(self): 

409 pass 

410 

411 @ATMT.timeout(WAIT_DATA, 1) 

412 def resend_ack(self): 

413 self.send(self.last_packet) 

414 raise self.WAIT_DATA() 

415 

416 @ATMT.receive_condition(WAIT_DATA) 

417 def receive_data(self, pkt): 

418 if TFTP_DATA in pkt: 

419 data = pkt[TFTP_DATA] 

420 if data.block == self.blk: 

421 raise self.DATA(data) 

422 

423 @ATMT.action(receive_data) 

424 def ack_data(self): 

425 self.last_packet = self.l3 / TFTP_ACK(block=self.blk % 65536) 

426 self.send(self.last_packet) 

427 

428 @ATMT.state() 

429 def DATA(self, data): 

430 self.filedata += data.load 

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

432 raise self.END() 

433 self.blk += 1 

434 raise self.WAIT_DATA() 

435 

436 @ATMT.state(final=1) 

437 def END(self): 

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

439 return self.filename, self.filedata 

440 

441 

442class TFTP_RRQ_server(Automaton): 

443 """ 

444 TFTP automaton to serve local files 

445 

446 You can't use 'store' and 'dir' at the same time. 

447 

448 :param store: (optional) a dictionary that contains the file data, like 

449 {"thefile": b"data"}. 

450 :param dir: (optional) a folder that contains the data file data. 

451 :param joker: (optional) data to return when no file/data is found. 

452 :param ip: (optional) the local IP to listen on. 

453 :param sport: (optional) the local port (by default: random) 

454 :param serve_one: (optional) close after serving one client (default: False) 

455 """ 

456 

457 MAX_RETRIES = 3 

458 

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

460 if "iface" not in kargs and ip: 

461 ip = str(Net(ip)) 

462 kargs["iface"] = conf.route.route(ip)[0] 

463 kargs.setdefault("session", IPSession()) 

464 Automaton.parse_args(self, **kargs) 

465 if store is None: 

466 store = {} 

467 if dir is not None: 

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

469 else: 

470 self.dir = None 

471 self.store = store 

472 self.joker = joker 

473 self.ip = ip 

474 self.sport = sport 

475 self.serve_one = serve_one 

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

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

478 

479 def master_filter(self, pkt): 

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

481 

482 @ATMT.state(initial=1) 

483 def WAIT_RRQ(self): 

484 self.blksize = 512 

485 self.blk = 0 

486 

487 @ATMT.receive_condition(WAIT_RRQ) 

488 def receive_rrq(self, pkt): 

489 if TFTP_RRQ in pkt: 

490 raise self.RECEIVED_RRQ(pkt) 

491 

492 @ATMT.state() 

493 def RECEIVED_RRQ(self, pkt): 

494 ip = pkt[IP] 

495 options = pkt.getlayer(TFTP_Options) 

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

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

498 self.blk = 1 

499 self.retry_block = self.blk 

500 self.retries = 0 

501 self.data = None 

502 if self.filename in self.store: 

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

504 elif self.dir is not None: 

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

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

507 try: 

508 with open(fn) as fd: 

509 self.data = fd.read() 

510 except IOError: 

511 pass 

512 if self.data is None: 

513 self.data = self.joker 

514 

515 if options: 

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

517 if opt: 

518 try: 

519 blksize = int(opt[0].value) 

520 if blksize < 8 or blksize > 65464: 

521 raise ValueError 

522 self.blksize = blksize 

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

524 except ValueError: 

525 opt = [] 

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

527 self.send(self.last_packet) 

528 

529 @ATMT.condition(RECEIVED_RRQ) 

530 def file_in_store(self): 

531 if self.data is not None: 

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

533 raise self.SEND_FILE() 

534 

535 @ATMT.condition(RECEIVED_RRQ) 

536 def file_not_found(self): 

537 if self.data is None: 

538 raise self.WAIT_RRQ() 

539 

540 @ATMT.action(file_not_found) 

541 def send_error(self): 

542 self.send(self.l3 / TFTP_ERROR( 

543 errorcode=1, 

544 errormsg=TFTP_Error_Codes[1], 

545 )) 

546 

547 @ATMT.state() 

548 def SEND_FILE(self): 

549 self.send( 

550 self.l3 / TFTP_DATA(block=self.blk % 65536) / 

551 self.data[(self.blk - 1) * self.blksize:self.blk * self.blksize] 

552 ) 

553 

554 @ATMT.timeout(SEND_FILE, 3) 

555 def timeout_waiting_ack(self): 

556 if self.retry_block != self.blk: 

557 self.retry_block = self.blk 

558 self.retries = 0 

559 if self.retries >= self.MAX_RETRIES: 

560 if self.serve_one: 

561 raise self.END() 

562 raise self.WAIT_RRQ() 

563 self.retries += 1 

564 raise self.SEND_FILE() 

565 

566 @ATMT.receive_condition(SEND_FILE) 

567 def received_ack(self, pkt): 

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

569 raise self.RECEIVED_ACK() 

570 

571 @ATMT.state() 

572 def RECEIVED_ACK(self): 

573 self.blk += 1 

574 

575 @ATMT.condition(RECEIVED_ACK) 

576 def no_more_data(self): 

577 if self.blk > self.blknb: 

578 if self.serve_one: 

579 raise self.END() 

580 raise self.WAIT_RRQ() 

581 

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

583 def data_remaining(self): 

584 raise self.SEND_FILE() 

585 

586 @ATMT.state(final=1) 

587 def END(self): 

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