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

1# SPDX-License-Identifier: GPL-2.0-or-later 

2# This file is part of Scapy 

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

4 

5# scapy.contrib.description = Internet Group Management Protocol v1/v2/v3 

6# scapy.contrib.status = loads 

7 

8""" 

9IGMP v1/v2/v3 as per RFC2236 and RFC9776 

10 

11Example: IGMPv1 Membership Report (=Join) 

12 

13 >>> send(IP() / IGMP_MR(gaddr="224.0.0.251")) 

14 

15Example: IGMPv3 Membership Report (=Join) 

16 

17 >>> send(IP() / IGMPv3_MR(records=[IGMPv3_MR_Group(rtype=4, maddr="224.0.0.251")])) 

18 

19Example: IGMPv1/v2 Membership Query 

20 

21 >>> sr(IP() / IGMP_MQ(gaddr="224.0.0.251"), multi=True) 

22 

23Example: IGMPv3 Membership Query 

24 

25 >>> sr(IP() / IGMPv3_MQ(gaddr="224.0.0.251"), multi=True) 

26 

27Example: IGMPv2 Leave 

28 

29 >>> send(IP() / IGMP_LG(gaddr="225.0.0.251")) 

30""" 

31 

32import struct 

33 

34from scapy.config import conf 

35from scapy.fields import ( 

36 BitField, 

37 ByteEnumField, 

38 ByteField, 

39 FieldLenField, 

40 FieldListField, 

41 FlagsField, 

42 IPField, 

43 PacketListField, 

44 ScalingField, 

45 ShortField, 

46 XShortField, 

47) 

48from scapy.packet import bind_layers, Packet, bind_top_down 

49from scapy.plist import PacketList 

50from scapy.sendrecv import sr, send 

51from scapy.utils import checksum 

52 

53from scapy.layers.inet import IP, IPOption_Router_Alert, DestIPField 

54 

55 

56class _MRCodeField(ScalingField): 

57 """ 

58 Max Resp Code field. 

59 

60 """ 

61 

62 def i2m(self, pkt, x): 

63 if x is None: 

64 if pkt.type == 0x11: # Membership Query 

65 return 20 

66 else: 

67 return 0 

68 if pkt.type == 0x11 and isinstance(pkt, IGMPv3): 

69 # IGMP v3 - RFC9776 sect 4.1.1 

70 if x < 128: 

71 return x 

72 else: 

73 exp = 0 

74 x >>= 3 

75 while x > 31: 

76 exp += 1 

77 x >>= 1 

78 exp <<= 4 

79 return 0x80 | exp | (x & 0x0F) 

80 return super(_MRCodeField, self).i2m(pkt, x) 

81 

82 def i2h(self, pkt, x): 

83 return super(_MRCodeField, self).i2h(pkt, self.i2m(pkt, x)) 

84 

85 def m2i(self, pkt, x): 

86 if pkt.type == 0x11 and isinstance(pkt, IGMPv3): 

87 # IGMP v3 - RFC9776 sect 4.1.1 

88 if x < 128: 

89 return x 

90 else: 

91 mant, exp = (x & 0x0F), (x >> 4) & 0x7 

92 return (mant | 0x10) << (exp + 3) 

93 return super(_MRCodeField, self).i2m(pkt, x) 

94 

95 

96class IGMP(Packet): 

97 """ 

98 General IGMP v1/v2 message. 

99 """ 

100 

101 fields_desc = [ 

102 ByteEnumField( 

103 "type", 

104 0x11, 

105 { 

106 0x11: "Group Membership Query", 

107 0x12: "Version 1 - Membership Report", 

108 0x16: "Version 2 - Membership Report", 

109 0x17: "Leave Group", 

110 }, 

111 ), 

112 _MRCodeField("mrcode", None, unit="1/10sec"), 

113 XShortField("chksum", None), 

114 IPField("gaddr", "0.0.0.0"), 

115 ] 

116 

117 def post_build(self, p, pay): 

118 p += pay 

119 if self.chksum is None: 

120 p = p[:2] + struct.pack("!H", checksum(p)) + p[4:] 

121 return p 

122 

123 def add_underlayer(self, underlayer): 

124 # Add option to parent IP layer 

125 if isinstance(underlayer, IP): 

126 if not underlayer.options: 

127 underlayer.options.append(IPOption_Router_Alert()) 

128 super(IGMP, self).add_underlayer(underlayer) 

129 

130 @classmethod 

131 def dispatch_hook(cls, _pkt=None, *args, **kargs): 

132 typ = None 

133 

134 # Get packet type from build or dissection 

135 if "type" in kargs: 

136 typ = kargs["type"] 

137 elif _pkt and len(_pkt) >= 4: 

138 typ = _pkt[0] 

139 

140 # Return the proper class depending on the packet type 

141 if typ is not None: 

142 if typ == 0x11: 

143 if (_pkt and len(_pkt) >= 12) or (not _pkt and issubclass(cls, IGMPv3)): 

144 return IGMPv3_MQ 

145 else: 

146 return IGMP_MQ 

147 elif typ == 0x12: 

148 return IGMP_MR 

149 elif typ == 0x16: 

150 return IGMPv2_MR 

151 elif typ == 0x17: 

152 return IGMPv2_LG 

153 elif typ == 0x22: 

154 return IGMPv3_MR 

155 elif typ == 0x30: 

156 return IGMPv3_MRA 

157 elif typ == 0x31: 

158 return IGMPv3_MRS 

159 elif typ == 0x32: 

160 return IGMPv3_MRT 

161 

162 return IGMP_MQ 

163 

164 def answers(self, other): 

165 if other.type == 0x11: 

166 return self.type in [0x12, 0x16, 0x22] 

167 elif other.type == 0x31: 

168 return self.type == 0x30 

169 return False 

170 

171 def mysummary(self): 

172 t = self.__class__.__name__ 

173 if isinstance(self.underlayer, IP): 

174 return self.underlayer.sprintf( 

175 f"IGMP: %IP.src% > %IP.dst% %{t}.type% %{t}.gaddr%" 

176 ) 

177 else: 

178 return self.sprintf(f"IGMP %{t}.type% %{t}.gaddr%") 

179 

180 

181class IGMP_MQ(IGMP): 

182 """ 

183 IGMPv1/v2 Membership Query 

184 """ 

185 

186 name = "IGMPv1/v2 Membership Query" 

187 match_subclass = True 

188 type = 0x11 

189 

190 

191class IGMPv2_LG(IGMP): 

192 """ 

193 IGMPv2 Leave Group 

194 """ 

195 

196 name = "IGMPv2 Leave Group" 

197 match_subclass = True 

198 type = 0x17 

199 

200 

201class IGMP_MR(IGMP): 

202 """ 

203 IGMPv1 Membership Report 

204 """ 

205 

206 name = "IGMPv1 Membership Report" 

207 type = 0x12 

208 match_subclass = True 

209 

210 

211class IGMPv2_MR(IGMP): 

212 """ 

213 IGMPv2 Membership Report 

214 """ 

215 

216 name = "IGMPv2 Membership Report" 

217 type = 0x16 

218 match_subclass = True 

219 

220 

221class IGMPv3(IGMP): 

222 """ 

223 IGMP Message Class for v3 

224 """ 

225 

226 name = "IGMPv3" 

227 

228 fields_desc = [ 

229 ByteEnumField( 

230 "type", 

231 0x11, 

232 { 

233 0x11: "Membership Query", 

234 0x22: "Version 3 - Membership Report", 

235 # RFC 4286 

236 0x30: "Multicast Router Advertisement", 

237 0x31: "Multicast Router Solicitation", 

238 0x32: "Multicast Router Termination", 

239 }, 

240 ), 

241 ] 

242 

243 @classmethod 

244 def dispatch_hook(cls, _pkt=None, *args, **kargs): 

245 return super(IGMPv3, cls).dispatch_hook(_pkt=_pkt, *args, **kargs) 

246 

247 def mysummary(self): 

248 """Display a summary of the IGMPv3 object.""" 

249 t = self.__class__.__name__ 

250 if isinstance(self.underlayer, IP): 

251 return self.underlayer.sprintf(f"IGMPv3: %IP.src% > %IP.dst% %{t}.type%") 

252 else: 

253 return self.sprintf(f"IGMPv3 %{t}.type%") 

254 

255 

256class IGMPv3_MQ(IGMPv3): 

257 """ 

258 IGMPv3 Membership Query 

259 """ 

260 

261 type = 0x11 

262 name = "IGMPv3 Membership Query" 

263 match_subclass = True 

264 

265 fields_desc = [ 

266 IGMPv3, 

267 _MRCodeField("mrcode", 20, unit="1/10sec"), 

268 XShortField("chksum", None), 

269 IPField("gaddr", "0.0.0.0"), 

270 FlagsField( 

271 "flags", 

272 0, 

273 4, 

274 { 

275 0x8: "extended", 

276 }, 

277 ), 

278 BitField("s", 0, 1), 

279 BitField("qrv", 0, 3), 

280 ByteField("qqic", 0), 

281 FieldLenField("numsrc", None, count_of="srcaddrs"), 

282 FieldListField( 

283 "srcaddrs", 

284 None, 

285 IPField("sa", "0.0.0.0"), 

286 count_from=lambda x: x.numsrc, 

287 ), 

288 ] 

289 

290 

291class IGMPv3_MR_Group(Packet): 

292 """ 

293 IGMP Group Record for IGMPv3 Membership Report 

294 

295 This class should be added in the records of an instantiation of class IGMPv3_MR. 

296 """ 

297 

298 name = "IGMPv3 Group Record" 

299 fields_desc = [ 

300 ByteEnumField( 

301 "rtype", 

302 4, 

303 { 

304 1: "Mode Is Include", 

305 2: "Mode Is Exclude", 

306 3: "Change To Include Mode", 

307 4: "Change To Exclude Mode", 

308 5: "Allow New Sources", 

309 6: "Block Old Sources", 

310 }, 

311 ), 

312 ByteField("auxdlen", 0), 

313 FieldLenField("numsrc", None, count_of="srcaddrs"), 

314 IPField("maddr", "0.0.0.0"), 

315 FieldListField( 

316 "srcaddrs", 

317 [], 

318 IPField("sa", "0.0.0.0"), 

319 count_from=lambda x: x.numsrc, 

320 ), 

321 ] 

322 

323 def mysummary(self): 

324 """Display a summary of the IGMPv3 group record.""" 

325 return self.sprintf( 

326 "IGMPv3 Group Record %IGMPv3gr.type% %IGMPv3gr.maddr%" 

327 ) # noqa: E501 

328 

329 def default_payload_class(self, payload): 

330 return conf.padding_layer 

331 

332 

333class IGMPv3_MR(IGMPv3): 

334 """ 

335 IGMP Membership Report extension for IGMPv3 

336 """ 

337 

338 type = 0x22 

339 name = "IGMPv3 Membership Report" 

340 match_subclass = True 

341 

342 fields_desc = [ 

343 IGMPv3, 

344 ByteField("reserved", 0), 

345 XShortField("chksum", None), 

346 FlagsField( 

347 "flags", 

348 0, 

349 16, 

350 { 

351 0x8000: "extended", 

352 }, 

353 ), 

354 FieldLenField("numgrp", None, count_of="records"), 

355 PacketListField( 

356 "records", 

357 [], 

358 IGMPv3_MR_Group, 

359 count_from=lambda x: x.numgrp, 

360 ), 

361 ] 

362 

363 

364class IGMPv3_MRA(IGMPv3): 

365 """ 

366 IGMP Multicast Router Advertisement per RFC4286 

367 """ 

368 

369 type = 0x30 

370 name = "IGMPv3_MRA" 

371 match_subclass = True 

372 

373 fields_desc = [ 

374 IGMPv3, 

375 ByteField("advIntvl", 0), 

376 XShortField("chksum", None), 

377 ShortField("qryIntvl", 0), 

378 ShortField("robust", 0), 

379 ] 

380 

381 

382class IGMPv3_MRS(IGMPv3): 

383 """ 

384 IGMP Multicast Router Solicitation per RFC4286 

385 """ 

386 

387 type = 0x31 

388 name = "IGMPv3_MRS" 

389 match_subclass = True 

390 

391 fields_desc = [ 

392 IGMPv3, 

393 ByteField("reserved", 0), 

394 XShortField("chksum", None), 

395 ] 

396 

397 

398class IGMPv3_MRT(IGMPv3): 

399 """ 

400 IGMP Multicast Router Termination per RFC4286 

401 """ 

402 

403 type = 0x31 

404 name = "IGMPv3_MRT" 

405 match_subclass = True 

406 

407 fields_desc = [ 

408 IGMPv3, 

409 ByteField("reserved", 0), 

410 XShortField("chksum", None), 

411 ] 

412 

413 

414bind_layers(IP, IGMP, proto=2) 

415bind_top_down(IP, IGMP, proto=2, ttl=1, tos=0xC0) 

416 

417 

418def _igmp_mq_addr(pkt): 

419 if pkt.gaddr == "0.0.0.0": 

420 # General Query 

421 return "224.0.0.1" 

422 else: 

423 return pkt.gaddr 

424 

425 

426DestIPField.bind_addr(IGMP_MQ, _igmp_mq_addr) 

427DestIPField.bind_addr(IGMPv3_MQ, _igmp_mq_addr) 

428DestIPField.bind_addr(IGMPv2_LG, "224.0.0.2") 

429DestIPField.bind_addr(IGMP_MR, lambda pkt: pkt.gaddr) 

430DestIPField.bind_addr(IGMPv2_MR, lambda pkt: pkt.gaddr) 

431DestIPField.bind_addr(IGMPv3_MR, "224.0.0.22") 

432 

433# RFC4286 

434DestIPField.bind_addr(IGMPv3_MRA, "224.0.0.106") 

435DestIPField.bind_addr(IGMPv3_MRT, "224.0.0.106") 

436DestIPField.bind_addr(IGMPv3_MRS, "224.0.0.2") 

437 

438 

439@conf.commands.register 

440def igmp_join(gaddr: str, version=2, psrc=None, iface=None): 

441 """ 

442 Send a IGMP Membership Report to join a multicast group 

443 

444 :param gaddr: the IPv4 of the group to join 

445 :param version: whether to use IGMPv1, IGMPv2 or IGMPv3. Default: both 2 and 3 

446 :param psrc: (optional) the source IP 

447 """ 

448 if version == 1: 

449 pkt = IP(src=psrc) / IGMP_MR(gaddr=gaddr) 

450 elif version == 2: 

451 pkt = IP(src=psrc) / IGMPv2_MR(gaddr=gaddr) 

452 elif version == 3: 

453 pkt = IP(src=psrc) / IGMPv3_MR(records=[IGMPv3_MR_Group(rtype=4, maddr=gaddr)]) 

454 send(pkt, iface=iface) 

455 

456 

457@conf.commands.register 

458def igmp_leave(gaddr: str, version=2, psrc=None, iface=None): 

459 """ 

460 Send a IGMP Leave Group to leave a multicast group 

461 

462 :param gaddr: the IPv4 of the group to leave 

463 :param psrc: (optional) the source IP 

464 """ 

465 if version == 1: 

466 raise ValueError("IGMPv1 does not include a mechanism to leave !") 

467 elif version == 2: 

468 pkt = IP(src=psrc) / IGMPv2_LG(gaddr=gaddr) 

469 elif version == 3: 

470 pkt = IP(src=psrc) / IGMPv3_MR(records=[IGMPv3_MR_Group(rtype=3, maddr=gaddr)]) 

471 send(pkt, iface=iface) 

472 

473 

474class IGMPMQResult(PacketList): 

475 def __init__( 

476 self, 

477 res=None, 

478 name="IGMP-MR", 

479 stats=None, 

480 ): 

481 PacketList.__init__(self, res, name, stats) 

482 

483 

484@conf.commands.register 

485def igmp_query(gaddr: str = None, version=2, timeout=2): 

486 """ 

487 Send/receive a Membership Query to get the members of a multicast group 

488 

489 :param gaddr: the IPv4 of the group to query 

490 

491 Example:: 

492 

493 >>> pkts = igmp_query("224.0.0.251") 

494 >>> pkts.show() 

495 

496 Example 2:: 

497 

498 >>> pkts = igmp_query("239.255.255.250", version=3) 

499 >>> pkts.show() 

500 """ 

501 if version == 1: 

502 pkt = IP() / IGMP_MQ(mrcode=0, gaddr=gaddr) 

503 elif version == 2: 

504 pkt = IP() / IGMP_MQ(mrcode=timeout * 10, gaddr=gaddr) 

505 elif version == 3: 

506 pkt = IP() / IGMPv3_MQ(mrcode=timeout * 10, gaddr=gaddr) 

507 

508 _old_checkIPaddr = conf.checkIPaddr 

509 conf.checkIPaddr = False 

510 try: 

511 return IGMPMQResult( 

512 [x.answer for x in sr(pkt, multi=True, timeout=timeout + 1)[0]] 

513 ) 

514 finally: 

515 conf.checkIPaddr = _old_checkIPaddr