Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/sessions.py: 16%

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

2# This file is part of Scapy 

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

4 

5""" 

6Sessions: decode flow of packets when sniffing 

7""" 

8 

9from collections import defaultdict 

10import socket 

11import struct 

12 

13from scapy.config import conf 

14from scapy.packet import Packet 

15from scapy.pton_ntop import inet_pton 

16 

17# Typing imports 

18from typing import ( 

19 Any, 

20 Callable, 

21 DefaultDict, 

22 Dict, 

23 Iterator, 

24 List, 

25 Optional, 

26 Tuple, 

27 Type, 

28 cast, 

29 TYPE_CHECKING, 

30) 

31from scapy.compat import Self 

32if TYPE_CHECKING: 

33 from scapy.supersocket import SuperSocket 

34 

35 

36class DefaultSession(object): 

37 """Default session: no stream decoding""" 

38 

39 def __init__(self, supersession: Optional[Self] = None): 

40 if supersession and not isinstance(supersession, DefaultSession): 

41 supersession = supersession() 

42 self.supersession = supersession 

43 

44 def process(self, pkt: Packet) -> Optional[Packet]: 

45 """ 

46 Called to pre-process the packet 

47 """ 

48 # Optionally handle supersession 

49 if self.supersession: 

50 return self.supersession.process(pkt) 

51 return pkt 

52 

53 def recv(self, sock: 'SuperSocket') -> Iterator[Packet]: 

54 """ 

55 Will be called by sniff() to ask for a packet 

56 """ 

57 pkt = sock.recv() 

58 if not pkt: 

59 return 

60 pkt = self.process(pkt) 

61 if pkt: 

62 yield pkt 

63 

64 

65class IPSession(DefaultSession): 

66 """Defragment IP packets 'on-the-flow'. 

67 

68 Usage: 

69 >>> sniff(session=IPSession) 

70 """ 

71 

72 def __init__(self, *args, **kwargs): 

73 # type: (*Any, **Any) -> None 

74 DefaultSession.__init__(self, *args, **kwargs) 

75 self.fragments = defaultdict(list) # type: DefaultDict[Tuple[Any, ...], List[Packet]] # noqa: E501 

76 

77 def process(self, packet: Packet) -> Optional[Packet]: 

78 from scapy.layers.inet import IP, _defrag_ip_pkt 

79 if not packet: 

80 return None 

81 if IP not in packet: 

82 return packet 

83 return _defrag_ip_pkt(packet, self.fragments)[1] # type: ignore 

84 

85 

86class StringBuffer(object): 

87 """StringBuffer is an object used to re-order data received during 

88 a TCP transmission. 

89 

90 Each TCP fragment contains a sequence number, which marks 

91 (relatively to the first sequence number) the index of the data contained 

92 in the fragment. 

93 

94 If a TCP fragment is missed, this class will fill the missing space with 

95 zeros. 

96 """ 

97 

98 def __init__(self): 

99 # type: () -> None 

100 self.content = bytearray(b"") 

101 self.content_len = 0 

102 self.noff = 0 # negative offset 

103 self.incomplete = [] # type: List[Tuple[int, int]] 

104 

105 def append(self, data: bytes, seq: Optional[int] = None) -> None: 

106 if not data: 

107 return 

108 data_len = len(data) 

109 if seq is None: 

110 seq = self.content_len 

111 seq = seq - 1 - self.noff 

112 if seq < 0: 

113 # Data is located before the start of the current buffer 

114 # (e.g. the first fragment was missing) 

115 self.content = bytearray(b"\x00" * (-seq)) + self.content 

116 self.content_len += (-seq) 

117 self.noff += seq 

118 seq = 0 

119 if seq + data_len > self.content_len: 

120 # Data is located after the end of the current buffer 

121 self.content += b"\x00" * (seq - self.content_len + data_len) 

122 # As data was missing, mark it. 

123 # self.incomplete.append((self.content_len, seq)) 

124 self.content_len = seq + data_len 

125 assert len(self.content) == self.content_len 

126 # XXX removes empty space marker. 

127 # for ifrag in self.incomplete: 

128 # if [???]: 

129 # self.incomplete.remove([???]) 

130 memoryview(self.content)[seq:seq + data_len] = data 

131 

132 def shiftleft(self, i: int) -> None: 

133 self.content = self.content[i:] 

134 self.content_len -= i 

135 

136 def full(self): 

137 # type: () -> bool 

138 # Should only be true when all missing data was filled up, 

139 # (or there never was missing data) 

140 return bool(self) 

141 

142 def clear(self): 

143 # type: () -> None 

144 self.__init__() # type: ignore 

145 

146 def __bool__(self): 

147 # type: () -> bool 

148 return bool(self.content_len) 

149 __nonzero__ = __bool__ 

150 

151 def __len__(self): 

152 # type: () -> int 

153 return self.content_len 

154 

155 def __bytes__(self): 

156 # type: () -> bytes 

157 return bytes(self.content) 

158 

159 def __str__(self): 

160 # type: () -> str 

161 return cast(str, self.__bytes__()) 

162 

163 

164def streamcls(cls: Type[Packet]) -> Callable[ 

165 [bytes, Dict[str, Any], Dict[str, Any]], 

166 Optional[Packet], 

167]: 

168 """ 

169 Wraps a class for use when dissecting streams. 

170 """ 

171 if hasattr(cls, "tcp_reassemble"): 

172 return cls.tcp_reassemble # type: ignore 

173 else: 

174 # There is no tcp_reassemble. Just dissect the packet 

175 return lambda data, *_: data and cls(data) 

176 

177 

178class TCPSession(IPSession): 

179 """A Session that reconstructs TCP streams. 

180 

181 NOTE: this has the same effect as wrapping a real socket.socket into StreamSocket, 

182 but for all concurrent TCP streams (can be used on pcaps or sniffed sessions). 

183 

184 NOTE: only protocols that implement a ``tcp_reassemble`` function will be processed 

185 by this session. Other protocols will not be reconstructed. 

186 

187 DEV: implement a class-function `tcp_reassemble` in your Packet class:: 

188 

189 @classmethod 

190 def tcp_reassemble(cls, data, metadata, session): 

191 # data = the reassembled data from the same request/flow 

192 # metadata = empty dictionary, that can be used to store data 

193 # during TCP reassembly 

194 # session = a dictionary proper to the bidirectional TCP session, 

195 # that can be used to store anything 

196 [...] 

197 # If the packet is available, return it. Otherwise don't. 

198 # Whenever you return a packet, the buffer will be discarded. 

199 return pkt 

200 # Otherwise, maybe store stuff in metadata, and return None, 

201 # as you need additional data. 

202 return None 

203 

204 For more details and a real example, see: 

205 https://scapy.readthedocs.io/en/latest/usage.html#how-to-use-tcpsession-to-defragment-tcp-packets 

206 

207 :param app: Whether the socket is on application layer = has no TCP 

208 layer. This is identical to StreamSocket so only use this if your 

209 underlying source of data isn't a socket.socket. 

210 """ 

211 

212 def __init__(self, app=False, *args, **kwargs): 

213 # type: (bool, *Any, **Any) -> None 

214 super(TCPSession, self).__init__(*args, **kwargs) 

215 self.app = app 

216 if app: 

217 self.data = StringBuffer() 

218 self.metadata = {} # type: Dict[str, Any] 

219 self.session = {} # type: Dict[str, Any] 

220 else: 

221 # The StringBuffer() is used to build a global 

222 # string from fragments and their seq nulber 

223 self.tcp_frags = defaultdict( 

224 lambda: (StringBuffer(), {}) 

225 ) # type: DefaultDict[bytes, Tuple[StringBuffer, Dict[str, Any]]] 

226 self.tcp_sessions = defaultdict( 

227 dict 

228 ) # type: DefaultDict[bytes, Dict[str, Any]] 

229 # Setup stopping dissection condition 

230 from scapy.layers.inet import TCP 

231 self.stop_dissection_after = TCP 

232 

233 def _get_ident(self, pkt, session=False): 

234 # type: (Packet, bool) -> bytes 

235 underlayer = pkt["TCP"].underlayer 

236 af = socket.AF_INET6 if "IPv6" in pkt else socket.AF_INET 

237 src = underlayer and inet_pton(af, underlayer.src) or b"" 

238 dst = underlayer and inet_pton(af, underlayer.dst) or b"" 

239 if session: 

240 # Bidirectional 

241 def xor(x, y): 

242 # type: (bytes, bytes) -> bytes 

243 return bytes(a ^ b for a, b in zip(x, y)) 

244 return struct.pack("!4sH", xor(src, dst), pkt.dport ^ pkt.sport) 

245 else: 

246 # Uni-directional 

247 return src + dst + struct.pack("!HH", pkt.dport, pkt.sport) 

248 

249 def _strip_padding(self, pkt: Packet) -> Optional[bytes]: 

250 """Strip the packet of any padding, and return the padding. 

251 """ 

252 if isinstance(pkt, conf.padding_layer): 

253 return cast(bytes, pkt.load) 

254 pad = pkt.getlayer(conf.padding_layer) 

255 if pad is not None and pad.underlayer is not None: 

256 # strip padding 

257 del pad.underlayer.payload 

258 return cast(bytes, pad.load) 

259 return None 

260 

261 def process(self, 

262 pkt: Packet, 

263 cls: Optional[Type[Packet]] = None) -> Optional[Packet]: 

264 """Process each packet: matches the TCP seq/ack numbers 

265 to follow the TCP streams, and orders the fragments. 

266 """ 

267 packet = None # type: Optional[Packet] 

268 if self.app: 

269 # Special mode: Application layer. Use on top of TCP 

270 self.data.append(bytes(pkt)) 

271 if cls is None and not isinstance(pkt, bytes): 

272 cls = pkt.__class__ 

273 if "tcp_reassemble" in self.metadata: 

274 tcp_reassemble = self.metadata["tcp_reassemble"] 

275 elif cls is not None: 

276 self.metadata["tcp_reassemble"] = tcp_reassemble = streamcls(cls) 

277 else: 

278 return None 

279 if self.data.full(): 

280 packet = tcp_reassemble( 

281 bytes(self.data), 

282 self.metadata, 

283 self.session, 

284 ) 

285 if packet: 

286 padding = self._strip_padding(packet) 

287 if padding: 

288 # There is remaining data for the next payload. 

289 self.data.shiftleft(len(self.data) - len(padding)) 

290 # Skip full-padding 

291 if isinstance(packet, conf.padding_layer): 

292 return None 

293 else: 

294 # No padding (data) left. Clear 

295 self.data.clear() 

296 self.metadata.clear() 

297 return packet 

298 return None 

299 

300 _pkt = super(TCPSession, self).process(pkt) 

301 if _pkt is None: 

302 return None 

303 else: # Python 3.8 := would be nice 

304 pkt = _pkt 

305 

306 from scapy.layers.inet import IP, TCP 

307 if not pkt: 

308 return None 

309 if TCP not in pkt: 

310 return pkt 

311 pay = pkt[TCP].payload 

312 new_data = pay.original 

313 # Match packets by a unique TCP identifier 

314 ident = self._get_ident(pkt) 

315 data, metadata = self.tcp_frags[ident] 

316 tcp_session = self.tcp_sessions[self._get_ident(pkt, True)] 

317 # Handle TCP sequence numbers 

318 seq = pkt[TCP].seq 

319 if "seq" not in metadata: 

320 metadata["seq"] = seq 

321 if "next_seq" in metadata and seq < metadata["next_seq"]: 

322 # Retransmitted data (that we already returned) 

323 new_data = new_data[metadata["next_seq"] - seq:] 

324 if not new_data: 

325 return None 

326 seq = metadata["next_seq"] 

327 # Let's guess which class is going to be used 

328 if "pay_class" not in metadata: 

329 metadata["pay_class"] = pay_class = pkt[TCP].guess_payload_class(new_data) 

330 metadata["tcp_reassemble"] = tcp_reassemble = streamcls(pay_class) 

331 else: 

332 tcp_reassemble = metadata["tcp_reassemble"] 

333 

334 if pay: 

335 # Get a relative sequence number for a storage purpose 

336 relative_seq = metadata.get("relative_seq", None) 

337 if relative_seq is None: 

338 relative_seq = metadata["relative_seq"] = seq - 1 

339 seq = seq - relative_seq 

340 # Add the data to the buffer 

341 data.append(new_data, seq) 

342 

343 # Check TCP FIN or TCP RESET 

344 if pkt[TCP].flags.F or pkt[TCP].flags.R: 

345 metadata["tcp_end"] = True 

346 elif not pay: 

347 # If there's no payload and the stream isn't ending, ignore. 

348 return pkt 

349 

350 # In case any app layer protocol requires it, 

351 # allow the parser to inspect TCP PSH flag 

352 if pkt[TCP].flags.P: 

353 metadata["tcp_psh"] = True 

354 # XXX TODO: check that no empty space is missing in the buffer. 

355 # XXX Currently, if a TCP fragment was missing, we won't notice it. 

356 if data.full(): 

357 # Reassemble using all previous packets 

358 metadata["original"] = pkt 

359 metadata["ident"] = ident 

360 packet = tcp_reassemble( 

361 bytes(data), 

362 metadata, 

363 tcp_session 

364 ) 

365 # Stack the result on top of the previous frames 

366 if packet: 

367 if "seq" in metadata: 

368 pkt[TCP].seq = metadata["seq"] 

369 # Clear TCP reassembly metadata 

370 metadata.clear() 

371 # Check for padding 

372 padding = self._strip_padding(packet) 

373 while padding: 

374 # There is remaining data for the next payload. 

375 full_length = data.content_len - len(padding) 

376 metadata["relative_seq"] = relative_seq + full_length 

377 data.shiftleft(full_length) 

378 # There might be a sub-payload hidden in the padding 

379 sub_packet = tcp_reassemble( 

380 bytes(data), 

381 metadata, 

382 tcp_session 

383 ) 

384 if sub_packet: 

385 packet /= sub_packet 

386 padding = self._strip_padding(sub_packet) 

387 else: 

388 break 

389 else: 

390 # No padding (data) left. Clear 

391 data.clear() 

392 del self.tcp_frags[ident] 

393 # Minimum next seq 

394 metadata["next_seq"] = pkt[TCP].seq + len(new_data) 

395 # Skip full-padding 

396 if isinstance(packet, conf.padding_layer): 

397 return None 

398 # Rebuild resulting packet 

399 if pay: 

400 pay.underlayer.remove_payload() 

401 if IP in pkt: 

402 pkt[IP].len = None 

403 pkt[IP].chksum = None 

404 pkt = pkt / packet 

405 pkt.wirelen = None 

406 return pkt 

407 return None 

408 

409 def recv(self, sock: 'SuperSocket') -> Iterator[Packet]: 

410 """ 

411 Will be called by sniff() to ask for a packet 

412 """ 

413 pkt = sock.recv(stop_dissection_after=self.stop_dissection_after) 

414 # Now handle TCP reassembly 

415 if self.app: 

416 while pkt is not None: 

417 pkt = self.process(pkt) 

418 if pkt: 

419 yield pkt 

420 # keep calling process as there might be more 

421 pkt = b"" # type: ignore 

422 else: 

423 pkt = self.process(pkt) # type: ignore 

424 if pkt: 

425 yield pkt 

426 return None