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 

202class NBNSQueryResponse(Packet): 

203 name = "NBNS query response" 

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

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

206 ByteField("NULL", 0), 

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

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

209 IntField("TTL", 0x493e0), 

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

211 PacketListField("ADDR_ENTRY", 

212 [NBNS_ADD_ENTRY()], NBNS_ADD_ENTRY, 

213 length_from=lambda pkt: pkt.RDLENGTH) 

214 ] 

215 

216 def mysummary(self): 

217 if not self.ADDR_ENTRY or \ 

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

219 return "NBNSQueryResponse" 

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

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

222 self.ADDR_ENTRY[0].NB_ADDRESS 

223 ) 

224 

225 def answers(self, other): 

226 return ( 

227 isinstance(other, NBNSQueryRequest) and 

228 other.QUESTION_NAME == self.RR_NAME 

229 ) 

230 

231 

232bind_layers(NBNSHeader, NBNSQueryResponse, # RD+AA 

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

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

235 bind_bottom_up(NBNSHeader, NBNSQueryResponse, 

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

237 

238 

239# Node Status Request 

240# RFC1002 sect 4.2.17 

241 

242class NBNSNodeStatusRequest(NBNSQueryRequest): 

243 name = "NBNS status request" 

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

245 QUESTION_TYPE = 0x21 

246 

247 def mysummary(self): 

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

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

250 ) 

251 

252 

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

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

255 

256 

257# Node Status Response 

258# RFC1002 sect 4.2.18 

259 

260class NBNSNodeStatusResponseService(Packet): 

261 name = "NBNS Node Status Response Service" 

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

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

264 0x03: "messenger service", 

265 0x20: "file server service", 

266 0x1b: "domain master browser", 

267 0x1c: "domain controller", 

268 0x1e: "browser election service" 

269 }), 

270 ByteField("NAME_FLAGS", 0x4), 

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

272 

273 def default_payload_class(self, payload): 

274 return conf.padding_layer 

275 

276 

277class NBNSNodeStatusResponse(Packet): 

278 name = "NBNS Node Status Response" 

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

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

281 ByteField("NULL", 0), 

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

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

284 IntField("TTL", 0), 

285 ShortField("RDLENGTH", 83), 

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

287 count_of="NODE_NAME"), 

288 PacketListField("NODE_NAME", 

289 [NBNSNodeStatusResponseService()], 

290 NBNSNodeStatusResponseService, 

291 count_from=lambda pkt: pkt.NUM_NAMES), 

292 SourceMACField("MAC_ADDRESS"), 

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

294 

295 def answers(self, other): 

296 return ( 

297 isinstance(other, NBNSNodeStatusRequest) and 

298 other.QUESTION_NAME == self.RR_NAME 

299 ) 

300 

301 

302bind_layers(NBNSHeader, NBNSNodeStatusResponse, 

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

304 

305 

306# Name Registration Request 

307# RFC1002 sect 4.2.2 

308 

309class NBNSRegistrationRequest(Packet): 

310 name = "NBNS registration request" 

311 fields_desc = [ 

312 NetBIOSNameField("QUESTION_NAME", "Windows"), 

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

314 ByteField("NULL", 0), 

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

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

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

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

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

320 IntField("TTL", 0), 

321 ShortField("RDLENGTH", 6), 

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

323 BitEnumField("OWNER_NODE_TYPE", 00, 2, 

324 _NETBIOS_OWNER_MODE_TYPES), 

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

326 IPField("NB_ADDRESS", "127.0.0.1") 

327 ] 

328 

329 def mysummary(self): 

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

331 

332 

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

334bind_layers(NBNSHeader, NBNSRegistrationRequest, 

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

336 

337 

338# Wait for Acknowledgement Response 

339# RFC1002 sect 4.2.16 

340 

341class NBNSWackResponse(Packet): 

342 name = "NBNS Wait for Acknowledgement Response" 

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

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

345 ByteField("NULL", 0), 

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

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

348 IntField("TTL", 2), 

349 ShortField("RDLENGTH", 2), 

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

351 

352 

353bind_layers(NBNSHeader, NBNSWackResponse, 

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

355 

356# NetBIOS DATAGRAM HEADER 

357 

358 

359class NBTDatagram(Packet): 

360 name = "NBT Datagram Packet" 

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

362 ByteField("Flags", 0x02), 

363 ShortField("ID", 0), 

364 IPField("SourceIP", "127.0.0.1"), 

365 ShortField("SourcePort", 138), 

366 ShortField("Length", None), 

367 ShortField("Offset", 0), 

368 NetBIOSNameField("SourceName", "windows"), 

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

370 ByteField("NULL1", 0), 

371 NetBIOSNameField("DestinationName", "windows"), 

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

373 ByteField("NULL2", 0)] 

374 

375 def post_build(self, pkt, pay): 

376 if self.Length is None: 

377 length = len(pay) + 68 

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

379 return pkt + pay 

380 

381 

382# SESSION SERVICE PACKETS 

383 

384 

385class NBTSession(Packet): 

386 name = "NBT Session Packet" 

387 MAXLENGTH = 0x3ffff 

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

389 0x81: "Session Request", 

390 0x82: "Positive Session Response", 

391 0x83: "Negative Session Response", 

392 0x84: "Retarget Session Response", 

393 0x85: "Session Keepalive"}), 

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

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

396 

397 def post_build(self, pkt, pay): 

398 if self.LENGTH is None: 

399 length = len(pay) & self.MAXLENGTH 

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

401 return pkt + pay 

402 

403 def extract_padding(self, s): 

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

405 

406 @classmethod 

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

408 if len(data) < 4: 

409 return None 

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

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

412 return cls(data) 

413 

414 

415bind_bottom_up(UDP, NBNSHeader, dport=137) 

416bind_bottom_up(UDP, NBNSHeader, sport=137) 

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

418 

419bind_bottom_up(UDP, NBTDatagram, dport=138) 

420bind_bottom_up(UDP, NBTDatagram, sport=138) 

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

422 

423bind_bottom_up(TCP, NBTSession, dport=445) 

424bind_bottom_up(TCP, NBTSession, sport=445) 

425bind_bottom_up(TCP, NBTSession, dport=139) 

426bind_bottom_up(TCP, NBTSession, sport=139) 

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

428 

429 

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

431 

432 

433@conf.commands.register 

434def nbns_resolve( 

435 qname: str, 

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

437 raw: bool = False, 

438 timeout: int = 3, 

439 **kwargs, 

440) -> List[str]: 

441 """ 

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

443 

444 :param qname: the name to query 

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

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

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

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

449 """ 

450 kwargs.setdefault("verbose", 0) 

451 

452 # Unify types (for caching) 

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

454 

455 # Check cache 

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

457 result = _nbns_cache.get(cache_ident) 

458 if result: 

459 return result 

460 

461 if iface is None: 

462 ifaces = [ 

463 x 

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

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

466 ] 

467 elif isinstance(iface, list): 

468 ifaces = iface 

469 else: 

470 ifaces = [iface] 

471 

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

473 requests = [] 

474 for iface in ifaces: 

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

476 if bdcst == "255.255.255.255": 

477 continue 

478 requests.append( 

479 IP(dst=bdcst) / 

480 UDP() / 

481 NBNSHeader() / 

482 NBNSQueryRequest(QUESTION_NAME=qname) 

483 ) 

484 

485 if not requests: 

486 return None 

487 

488 # Perform requests, get the first response 

489 try: 

490 old_checkIPAddr = conf.checkIPaddr 

491 conf.checkIPaddr = False 

492 

493 res = sr1( 

494 requests, 

495 timeout=timeout, 

496 first=True, 

497 **kwargs, 

498 ) 

499 finally: 

500 conf.checkIPaddr = old_checkIPAddr 

501 

502 if res is not None: 

503 if raw: 

504 # Raw 

505 result = res 

506 else: 

507 # Get IP 

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

509 if result: 

510 # Cache it 

511 _nbns_cache[cache_ident] = result 

512 return result 

513 else: 

514 raise TimeoutError 

515 

516 

517class NBNS_am(AnsweringMachine): 

518 function_name = "nbnsd" 

519 filter = "udp port 137" 

520 sniff_options = {"store": 0} 

521 

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

523 """ 

524 NBNS answering machine 

525 

526 :param server_name: the netbios server name to match 

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

528 :param ip: the IP to answer with 

529 """ 

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

531 self.ip = ip 

532 if isinstance(from_ip, str): 

533 self.from_ip = Net(from_ip) 

534 else: 

535 self.from_ip = from_ip 

536 

537 def is_request(self, req): 

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

539 return False 

540 return NBNSQueryRequest in req and ( 

541 not self.ServerName or 

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

543 ) 

544 

545 def make_reply(self, req): 

546 # type: (Packet) -> Packet 

547 resp = Ether( 

548 dst=req[Ether].src, 

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

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

551 sport=req.dport, 

552 dport=req.sport, 

553 ) 

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

555 resp /= NBNSHeader() / NBNSQueryResponse( 

556 RR_NAME=self.ServerName or req.QUESTION_NAME, 

557 SUFFIX=req.SUFFIX, 

558 ADDR_ENTRY=[NBNS_ADD_ENTRY(NB_ADDRESS=address)] 

559 ) 

560 resp.NAME_TRN_ID = req.NAME_TRN_ID 

561 return resp