(cls, data, metadata, session)
| 1191 | |
| 1192 | @classmethod |
| 1193 | def tcp_reassemble(cls, data, metadata, session): |
| 1194 | # Used with TCPSession |
| 1195 | from scapy.layers.tls.record import TLS |
| 1196 | from scapy.layers.tls.record_tls13 import TLS13 |
| 1197 | if cls in (TLS, TLS13): |
| 1198 | length = struct.unpack("!H", data[3:5])[0] + 5 |
| 1199 | if len(data) >= length: |
| 1200 | # get the underlayer as it is used to populate tls_session |
| 1201 | if "original" not in metadata: |
| 1202 | return cls(data) |
| 1203 | underlayer = metadata["original"][TCP].copy() |
| 1204 | underlayer.remove_payload() |
| 1205 | # eventually get the tls_session now for TLS.dispatch_hook |
| 1206 | tls_session = None |
| 1207 | if conf.tls_session_enable: |
| 1208 | s = tlsSession() |
| 1209 | s.set_underlayer(underlayer) |
| 1210 | tls_session = conf.tls_sessions.find(s) |
| 1211 | if tls_session: |
| 1212 | if tls_session.dport != underlayer.dport: |
| 1213 | tls_session = tls_session.mirror() |
| 1214 | if tls_session.firsttcp == underlayer.seq: |
| 1215 | log_runtime.info( |
| 1216 | "TLS: session %s is a duplicate of a previous " |
| 1217 | "dissection. Discard it" % repr(tls_session) |
| 1218 | ) |
| 1219 | conf.tls_sessions.rem(tls_session, force=True) |
| 1220 | tls_session = None |
| 1221 | return cls(data, _underlayer=underlayer, tls_session=tls_session) |
| 1222 | else: |
| 1223 | return cls(data) |
| 1224 | |
| 1225 | |
| 1226 | ############################################################################### |
nothing calls this directly
no test coverage detected