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

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""" 

7NetBIOS over TCP/IP 

8 

9[RFC 1001/1002] 

10""" 

11 

12import struct 

13from scapy.arch import get_if_addr 

14from scapy.base_classes import Net 

15from scapy.ansmachine import AnsweringMachine 

16from scapy.compat import bytes_encode 

17from scapy.config import conf 

18 

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

20from scapy.fields import ( 

21 BitEnumField, 

22 BitField, 

23 ByteEnumField, 

24 ByteField, 

25 FieldLenField, 

26 FlagsField, 

27 IPField, 

28 IntField, 

29 NetBIOSNameField, 

30 PacketListField, 

31 ShortEnumField, 

32 ShortField, 

33 StrFixedLenField, 

34 XShortField, 

35 XStrFixedLenField 

36) 

37from scapy.interfaces import _GlobInterfaceType 

38from scapy.sendrecv import sr1 

39from scapy.layers.inet import IP, UDP, TCP 

40from scapy.layers.l2 import Ether, SourceMACField 

41 

42# Typing imports 

43from typing import ( 

44 List, 

45 Union, 

46) 

47 

48 

49class NetBIOS_DS(Packet): 

50 name = "NetBIOS datagram service" 

51 fields_desc = [ 

52 ByteEnumField("type", 17, {17: "direct_group"}), 

53 ByteField("flags", 0), 

54 XShortField("id", 0), 

55 IPField("src", "127.0.0.1"), 

56 ShortField("sport", 138), 

57 ShortField("len", None), 

58 ShortField("ofs", 0), 

59 NetBIOSNameField("srcname", ""), 

60 NetBIOSNameField("dstname", ""), 

61 ] 

62 

63 def post_build(self, p, pay): 

64 p += pay 

65 if self.len is None: 

66 tmp_len = len(p) - 14 

67 p = p[:10] + struct.pack("!H", tmp_len) + p[12:] 

68 return p 

69 

70# ShortField("length",0), 

71# ShortField("Delimiter",0), 

72# ByteField("command",0), 

73# ByteField("data1",0), 

74# ShortField("data2",0), 

75# ShortField("XMIt",0), 

76# ShortField("RSPCor",0), 

77# StrFixedLenField("dest","",16), 

78# StrFixedLenField("source","",16), 

79# 

80# ] 

81# 

82 

83# NetBIOS 

84 

85 

86_NETBIOS_SUFFIXES = { 

87 0x4141 + 0x00: "Workstation Service", 

88 0x4141 + 0x01: "Messenger Service", 

89 0x4141 + 0x03: "Messenger service", 

90 0x4141 + 0x06: "RAS Server Service", 

91 0x4141 + 0x1B: "Exchange MTA", 

92 0x4141 + 0x1F: "NetDDE Service", 

93 0x4141 + 0x20: "File Server Service", 

94 0x4141 + 0x21: "RAS Client Service", 

95 0x4141 + 0x22: "Exchange Interchange Service", 

96 0x4141 + 0x23: "Exchange Store", 

97 0x4141 + 0x24: "Exchange Directory", 

98 0x4141 + 0x30: "Modern Sharing Server Service", 

99 0x4141 + 0x31: "Modern Sharing Client Service", 

100 0x4141 + 0x43: "SMS Client Remote Control", 

101 0x4141 + 0x44: "SMS Admin Remote Control Tool", 

102 0x4141 + 0x45: "SMS Client Remote Chat", 

103 0x4141 + 0x46: "SMS Client Remote Transfer", 

104 0x4141 + 0x4C: "DEC Pathworks TCP/IP Service", 

105 0x4141 + 0x52: "DEC Pathworks TCP/IP Service", 

106 0x4141 + 0x6A: "Exchange IMC", 

107 0x4141 + 0x87: "Exchange MTA", 

108 0x4141 + 0xBE: "Network Monitor Agent", 

109 0x4141 + 0xBF: "Network Monitor Apps", 

110 0x4141 + 0x10b: "Domain Master Browser", 

111 0x4141 + 0x10c: "Domain Controller", 

112 0x4141 + 0x10e: "Browser Election Service", 

113 0x4141 + 0x200: "File Server Service", 

114} 

115 

116_NETBIOS_QRTYPES = { 

117 0x20: "NB", 

118 0x21: "NBSTAT" 

119} 

120 

121_NETBIOS_QRCLASS = { 

122 1: "INTERNET" 

123} 

124 

125_NETBIOS_RNAMES = { 

126 0xC00C: "Label String Pointer to QUESTION_NAME" 

127} 

128 

129_NETBIOS_OWNER_MODE_TYPES = { 

130 0: "B node", 

131 1: "P node", 

132 2: "M node", 

133 3: "H node" 

134} 

135 

136_NETBIOS_GNAMES = { 

137 0: "Unique name", 

138 1: "Group name" 

139} 

140 

141 

142class NBNSHeader(Packet): 

143 name = "NBNS Header" 

144 fields_desc = [ 

145 ShortField("NAME_TRN_ID", 0), 

146 BitField("RESPONSE", 0, 1), 

147 BitField("OPCODE", 0, 4), 

148 FlagsField("NM_FLAGS", 0, 7, ["B", 

149 "res1", 

150 "res0", 

151 "RA", 

152 "RD", 

153 "TC", 

154 "AA"]), 

155 BitField("RCODE", 0, 4), 

156 ShortField("QDCOUNT", 0), 

157 ShortField("ANCOUNT", 0), 

158 ShortField("NSCOUNT", 0), 

159 ShortField("ARCOUNT", 0), 

160 ] 

161 

162 def hashret(self): 

163 return b"NBNS" + struct.pack("!B", self.OPCODE) 

164 

165 

166# Name Query Request 

167# RFC1002 sect 4.2.12 

168 

169 

170class NBNSQueryRequest(Packet): 

171 name = "NBNS query request" 

172 fields_desc = [NetBIOSNameField("QUESTION_NAME", "windows"), 

173 ShortEnumField("SUFFIX", 0x4141, _NETBIOS_SUFFIXES), 

174 ByteField("NULL", 0), 

175 ShortEnumField("QUESTION_TYPE", 0x20, _NETBIOS_QRTYPES), 

176 ShortEnumField("QUESTION_CLASS", 1, _NETBIOS_QRCLASS)] 

177 

178 def mysummary(self): 

179 return "NBNSQueryRequest who has '\\\\%s'" % ( 

180 self.QUESTION_NAME.decode(errors="backslashreplace") 

181 ) 

182 

183 

184bind_layers(NBNSHeader, NBNSQueryRequest, 

185 OPCODE=0x0, NM_FLAGS=0x11, QDCOUNT=1) 

186 

187 

188# Name Query Response 

189# RFC1002 sect 4.2.13 

190 

191 

192class NBNS_ADD_ENTRY(Packet): 

193 fields_desc = [ 

194 BitEnumField("G", 0, 1, _NETBIOS_GNAMES), 

195 BitEnumField("OWNER_NODE_TYPE", 00, 2, 

196 _NETBIOS_OWNER_MODE_TYPES), 

197 BitEnumField("UNUSED", 0, 13, {0: "Unused"}), 

198 IPField("NB_ADDRESS", "127.0.0.1") 

199 ] 

200 

201 def default_payload_class(self, payload): 

202 return conf.padding_layer 

203 

204 

205class NBNSQueryResponse(Packet): 

206 name = "NBNS query response" 

207 fields_desc = [NetBIOSNameField("RR_NAME", "windows"), 

208 ShortEnumField("SUFFIX", 0x4141, _NETBIOS_SUFFIXES), 

209 ByteField("NULL", 0), 

210 ShortEnumField("QUESTION_TYPE", 0x20, _NETBIOS_QRTYPES), 

211 ShortEnumField("QUESTION_CLASS", 1, _NETBIOS_QRCLASS), 

212 IntField("TTL", 0x493e0), 

213 FieldLenField("RDLENGTH", None, length_of="ADDR_ENTRY"), 

214 PacketListField("ADDR_ENTRY", 

215 [NBNS_ADD_ENTRY()], NBNS_ADD_ENTRY, 

216 length_from=lambda pkt: pkt.RDLENGTH) 

217 ] 

218 

219 def mysummary(self): 

220 if not self.ADDR_ENTRY or \ 

221 not isinstance(self.ADDR_ENTRY[0], NBNS_ADD_ENTRY): 

222 return "NBNSQueryResponse" 

223 return "NBNSQueryResponse '\\\\%s' is at %s" % ( 

224 self.RR_NAME.decode(errors="backslashreplace"), 

225 self.ADDR_ENTRY[0].NB_ADDRESS 

226 ) 

227 

228 def answers(self, other): 

229 return ( 

230 isinstance(other, NBNSQueryRequest) and 

231 other.QUESTION_NAME == self.RR_NAME 

232 ) 

233 

234 

235bind_layers(NBNSHeader, NBNSQueryResponse, # RD+AA 

236 OPCODE=0x0, NM_FLAGS=0x50, RESPONSE=1, ANCOUNT=1) 

237for _flg in [0x58, 0x70, 0x78]: 

238 bind_bottom_up(NBNSHeader, NBNSQueryResponse, 

239 OPCODE=0x0, NM_FLAGS=_flg, RESPONSE=1, ANCOUNT=1) 

240 

241 

242# Node Status Request 

243# RFC1002 sect 4.2.17 

244 

245class NBNSNodeStatusRequest(NBNSQueryRequest): 

246 name = "NBNS status request" 

247 QUESTION_NAME = b"*" + b"\x00" * 14 

248 QUESTION_TYPE = 0x21 

249 

250 def mysummary(self): 

251 return "NBNSNodeStatusRequest who has '\\\\%s'" % ( 

252 self.QUESTION_NAME.decode(errors="backslashreplace") 

253 ) 

254 

255 

256bind_bottom_up(NBNSHeader, NBNSNodeStatusRequest, OPCODE=0x0, NM_FLAGS=0, QDCOUNT=1) 

257bind_layers(NBNSHeader, NBNSNodeStatusRequest, OPCODE=0x0, NM_FLAGS=1, QDCOUNT=1) 

258 

259 

260# Node Status Response 

261# RFC1002 sect 4.2.18 

262 

263class NBNSNodeStatusResponseService(Packet): 

264 name = "NBNS Node Status Response Service" 

265 fields_desc = [StrFixedLenField("NETBIOS_NAME", "WINDOWS ", 15), 

266 ByteEnumField("SUFFIX", 0, {0: "workstation", 

267 0x03: "messenger service", 

268 0x20: "file server service", 

269 0x1b: "domain master browser", 

270 0x1c: "domain controller", 

271 0x1e: "browser election service" 

272 }), 

273 ByteField("NAME_FLAGS", 0x4), 

274 ByteEnumField("UNUSED", 0, {0: "unused"})] 

275 

276 def default_payload_class(self, payload): 

277 return conf.padding_layer 

278 

279 

280class NBNSNodeStatusResponse(Packet): 

281 name = "NBNS Node Status Response" 

282 fields_desc = [NetBIOSNameField("RR_NAME", "windows"), 

283 ShortEnumField("SUFFIX", 0x4141, _NETBIOS_SUFFIXES), 

284 ByteField("NULL", 0), 

285 ShortEnumField("RR_TYPE", 0x21, _NETBIOS_QRTYPES), 

286 ShortEnumField("RR_CLASS", 1, _NETBIOS_QRCLASS), 

287 IntField("TTL", 0), 

288 ShortField("RDLENGTH", 83), 

289 FieldLenField("NUM_NAMES", None, fmt="B", 

290 count_of="NODE_NAME"), 

291 PacketListField("NODE_NAME", 

292 [NBNSNodeStatusResponseService()], 

293 NBNSNodeStatusResponseService, 

294 count_from=lambda pkt: pkt.NUM_NAMES), 

295 SourceMACField("MAC_ADDRESS"), 

296 XStrFixedLenField("STATISTICS", b"", 46)] 

297 

298 def answers(self, other): 

299 return ( 

300 isinstance(other, NBNSNodeStatusRequest) and 

301 other.QUESTION_NAME == self.RR_NAME 

302 ) 

303 

304 

305bind_layers(NBNSHeader, NBNSNodeStatusResponse, 

306 OPCODE=0x0, NM_FLAGS=0x40, RESPONSE=1, ANCOUNT=1) 

307 

308 

309# Name Registration Request 

310# RFC1002 sect 4.2.2 

311 

312class NBNSRegistrationRequest(Packet): 

313 name = "NBNS registration request" 

314 fields_desc = [ 

315 NetBIOSNameField("QUESTION_NAME", "Windows"), 

316 ShortEnumField("SUFFIX", 0x4141, _NETBIOS_SUFFIXES), 

317 ByteField("NULL", 0), 

318 ShortEnumField("QUESTION_TYPE", 0x20, _NETBIOS_QRTYPES), 

319 ShortEnumField("QUESTION_CLASS", 1, _NETBIOS_QRCLASS), 

320 ShortEnumField("RR_NAME", 0xC00C, _NETBIOS_RNAMES), 

321 ShortEnumField("RR_TYPE", 0x20, _NETBIOS_QRTYPES), 

322 ShortEnumField("RR_CLASS", 1, _NETBIOS_QRCLASS), 

323 IntField("TTL", 0), 

324 ShortField("RDLENGTH", 6), 

325 BitEnumField("G", 0, 1, _NETBIOS_GNAMES), 

326 BitEnumField("OWNER_NODE_TYPE", 00, 2, 

327 _NETBIOS_OWNER_MODE_TYPES), 

328 BitEnumField("UNUSED", 0, 13, {0: "Unused"}), 

329 IPField("NB_ADDRESS", "127.0.0.1") 

330 ] 

331 

332 def mysummary(self): 

333 return self.sprintf("Register %G% %QUESTION_NAME% at %NB_ADDRESS%") 

334 

335 

336bind_bottom_up(NBNSHeader, NBNSRegistrationRequest, OPCODE=0x5) 

337bind_layers(NBNSHeader, NBNSRegistrationRequest, 

338 OPCODE=0x5, NM_FLAGS=0x11, QDCOUNT=1, ARCOUNT=1) 

339 

340 

341# Wait for Acknowledgement Response 

342# RFC1002 sect 4.2.16 

343 

344class NBNSWackResponse(Packet): 

345 name = "NBNS Wait for Acknowledgement Response" 

346 fields_desc = [NetBIOSNameField("RR_NAME", "windows"), 

347 ShortEnumField("SUFFIX", 0x4141, _NETBIOS_SUFFIXES), 

348 ByteField("NULL", 0), 

349 ShortEnumField("RR_TYPE", 0x20, _NETBIOS_QRTYPES), 

350 ShortEnumField("RR_CLASS", 1, _NETBIOS_QRCLASS), 

351 IntField("TTL", 2), 

352 ShortField("RDLENGTH", 2), 

353 BitField("RDATA", 10512, 16)] # 10512=0010100100010000 

354 

355 

356bind_layers(NBNSHeader, NBNSWackResponse, 

357 OPCODE=0x7, NM_FLAGS=0x40, RESPONSE=1, ANCOUNT=1) 

358 

359# NetBIOS DATAGRAM HEADER 

360 

361 

362class NBTDatagram(Packet): 

363 name = "NBT Datagram Packet" 

364 fields_desc = [ByteField("Type", 0x10), 

365 ByteField("Flags", 0x02), 

366 ShortField("ID", 0), 

367 IPField("SourceIP", "127.0.0.1"), 

368 ShortField("SourcePort", 138), 

369 ShortField("Length", None), 

370 ShortField("Offset", 0), 

371 NetBIOSNameField("SourceName", "windows"), 

372 ShortEnumField("SUFFIX1", 0x4141, _NETBIOS_SUFFIXES), 

373 ByteField("NULL1", 0), 

374 NetBIOSNameField("DestinationName", "windows"), 

375 ShortEnumField("SUFFIX2", 0x4141, _NETBIOS_SUFFIXES), 

376 ByteField("NULL2", 0)] 

377 

378 def post_build(self, pkt, pay): 

379 if self.Length is None: 

380 length = len(pay) + 68 

381 pkt = pkt[:10] + struct.pack("!H", length) + pkt[12:] 

382 return pkt + pay 

383 

384 

385# SESSION SERVICE PACKETS 

386 

387 

388class NBTSession(Packet): 

389 name = "NBT Session Packet" 

390 MAXLENGTH = 0x3ffff 

391 fields_desc = [ByteEnumField("TYPE", 0, {0x00: "Session Message", 

392 0x81: "Session Request", 

393 0x82: "Positive Session Response", 

394 0x83: "Negative Session Response", 

395 0x84: "Retarget Session Response", 

396 0x85: "Session Keepalive"}), 

397 BitField("RESERVED", 0x00, 7), 

398 BitField("LENGTH", None, 17)] 

399 

400 def post_build(self, pkt, pay): 

401 if self.LENGTH is None: 

402 length = len(pay) & self.MAXLENGTH 

403 pkt = pkt[:1] + struct.pack("!I", length)[1:] 

404 return pkt + pay 

405 

406 def extract_padding(self, s): 

407 return s[:self.LENGTH], s[self.LENGTH:] 

408 

409 @classmethod 

410 def tcp_reassemble(cls, data, *args, **kwargs): 

411 if len(data) < 4: 

412 return None 

413 length = struct.unpack("!I", data[:4])[0] & cls.MAXLENGTH 

414 if len(data) >= length + 4: 

415 return cls(data) 

416 

417 

418bind_bottom_up(UDP, NBNSHeader, dport=137) 

419bind_bottom_up(UDP, NBNSHeader, sport=137) 

420bind_top_down(UDP, NBNSHeader, sport=137, dport=137) 

421 

422bind_bottom_up(UDP, NBTDatagram, dport=138) 

423bind_bottom_up(UDP, NBTDatagram, sport=138) 

424bind_top_down(UDP, NBTDatagram, sport=138, dport=138) 

425 

426bind_bottom_up(TCP, NBTSession, dport=445) 

427bind_bottom_up(TCP, NBTSession, sport=445) 

428bind_bottom_up(TCP, NBTSession, dport=139) 

429bind_bottom_up(TCP, NBTSession, sport=139) 

430bind_layers(TCP, NBTSession, dport=139, sport=139) 

431 

432 

433_nbns_cache = conf.netcache.new_cache("nbns_cache", 300) 

434 

435 

436@conf.commands.register 

437def nbns_resolve( 

438 qname: str, 

439 iface: Union[_GlobInterfaceType, List[_GlobInterfaceType]] = None, 

440 raw: bool = False, 

441 timeout: int = 3, 

442 **kwargs, 

443) -> List[str]: 

444 """ 

445 Perform a simple NBNS (NetBios Name Services) resolution with caching 

446 

447 :param qname: the name to query 

448 :param iface: the interfaces to use. (default: all) 

449 :param raw: return the whole netbios packet (default False) 

450 :param timeout: seconds until timeout (per server) 

451 :raise TimeoutError: if no DNS servers were reached in time. 

452 """ 

453 kwargs.setdefault("verbose", 0) 

454 

455 # Unify types (for caching) 

456 qname = NBNSQueryRequest.QUESTION_NAME.any2i(None, qname) 

457 

458 # Check cache 

459 cache_ident = qname + b"raw" if raw else b"" 

460 result = _nbns_cache.get(cache_ident) 

461 if result: 

462 return result 

463 

464 if iface is None: 

465 ifaces = [ 

466 x 

467 for name, x in conf.ifaces.items() 

468 if x.is_valid() and name != conf.loopback_name 

469 ] 

470 elif isinstance(iface, list): 

471 ifaces = iface 

472 else: 

473 ifaces = [iface] 

474 

475 # Builds a request for each broadcast address of each interface 

476 requests = [] 

477 for iface in ifaces: 

478 for bdcst in conf.route.get_if_bcast(iface): 

479 if bdcst == "255.255.255.255": 

480 continue 

481 requests.append( 

482 IP(dst=bdcst) / 

483 UDP() / 

484 NBNSHeader() / 

485 NBNSQueryRequest(QUESTION_NAME=qname) 

486 ) 

487 

488 if not requests: 

489 return None 

490 

491 # Perform requests, get the first response 

492 try: 

493 old_checkIPAddr = conf.checkIPaddr 

494 conf.checkIPaddr = False 

495 

496 res = sr1( 

497 requests, 

498 timeout=timeout, 

499 first=True, 

500 **kwargs, 

501 ) 

502 finally: 

503 conf.checkIPaddr = old_checkIPAddr 

504 

505 if res is not None: 

506 if raw: 

507 # Raw 

508 result = res 

509 else: 

510 # Get IP 

511 result = [x.NB_ADDRESS for x in res.ADDR_ENTRY] 

512 if result: 

513 # Cache it 

514 _nbns_cache[cache_ident] = result 

515 return result 

516 else: 

517 raise TimeoutError 

518 

519 

520class NBNS_am(AnsweringMachine): 

521 function_name = "nbnsd" 

522 filter = "udp port 137" 

523 sniff_options = {"store": 0} 

524 

525 def parse_options(self, server_name=None, from_ip=None, ip=None): 

526 """ 

527 NBNS answering machine 

528 

529 :param server_name: the netbios server name to match 

530 :param from_ip: an IP (can have a netmask) to filter on 

531 :param ip: the IP to answer with 

532 """ 

533 self.ServerName = bytes_encode(server_name or "") 

534 self.ip = ip 

535 if isinstance(from_ip, str): 

536 self.from_ip = Net(from_ip) 

537 else: 

538 self.from_ip = from_ip 

539 

540 def is_request(self, req): 

541 if self.from_ip and IP in req and req[IP].src not in self.from_ip: 

542 return False 

543 return NBNSQueryRequest in req and ( 

544 not self.ServerName or 

545 req[NBNSQueryRequest].QUESTION_NAME.strip() == self.ServerName 

546 ) 

547 

548 def make_reply(self, req): 

549 # type: (Packet) -> Packet 

550 resp = Ether( 

551 dst=req[Ether].src, 

552 src=None if req[Ether].dst == "ff:ff:ff:ff:ff:ff" else req[Ether].dst, 

553 ) / IP(dst=req[IP].src) / UDP( 

554 sport=req.dport, 

555 dport=req.sport, 

556 ) 

557 address = self.ip or get_if_addr(self.optsniff.get("iface", conf.iface)) 

558 resp /= NBNSHeader() / NBNSQueryResponse( 

559 RR_NAME=self.ServerName or req.QUESTION_NAME, 

560 SUFFIX=req.SUFFIX, 

561 ADDR_ENTRY=[NBNS_ADD_ENTRY(NB_ADDRESS=address)] 

562 ) 

563 resp.NAME_TRN_ID = req.NAME_TRN_ID 

564 return resp