A stream PCAP writer with more control than wrpcap()
| 2262 | |
| 2263 | |
| 2264 | class RawPcapWriter(GenericRawPcapWriter): |
| 2265 | """A stream PCAP writer with more control than wrpcap()""" |
| 2266 | |
| 2267 | def __init__(self, |
| 2268 | filename, # type: Union[IO[bytes], str] |
| 2269 | linktype=None, # type: Optional[int] |
| 2270 | gz=False, # type: bool |
| 2271 | endianness="", # type: str |
| 2272 | append=False, # type: bool |
| 2273 | sync=False, # type: bool |
| 2274 | nano=False, # type: bool |
| 2275 | snaplen=MTU, # type: int |
| 2276 | bufsz=4096, # type: int |
| 2277 | ): |
| 2278 | # type: (...) -> None |
| 2279 | """ |
| 2280 | :param filename: the name of the file to write packets to, or an open, |
| 2281 | writable file-like object. |
| 2282 | :param linktype: force linktype to a given value. If None, linktype is |
| 2283 | taken from the first writer packet |
| 2284 | :param gz: compress the capture on the fly |
| 2285 | :param endianness: force an endianness (little:"<", big:">"). |
| 2286 | Default is native |
| 2287 | :param append: append packets to the capture file instead of |
| 2288 | truncating it |
| 2289 | :param sync: do not bufferize writes to the capture file |
| 2290 | :param nano: use nanosecond-precision (requires libpcap >= 1.5.0) |
| 2291 | |
| 2292 | """ |
| 2293 | |
| 2294 | if linktype: |
| 2295 | self.linktype = linktype |
| 2296 | self.snaplen = snaplen |
| 2297 | self.append = append |
| 2298 | self.gz = gz |
| 2299 | self.endian = endianness |
| 2300 | self.sync = sync |
| 2301 | self.nano = nano |
| 2302 | if sync: |
| 2303 | bufsz = 0 |
| 2304 | |
| 2305 | if isinstance(filename, str): |
| 2306 | self.filename = filename |
| 2307 | if gz: |
| 2308 | self.f = cast(_ByteStream, gzip.open( |
| 2309 | filename, append and "ab" or "wb", 9 |
| 2310 | )) |
| 2311 | else: |
| 2312 | self.f = open(filename, append and "ab" or "wb", bufsz) |
| 2313 | else: |
| 2314 | self.f = filename |
| 2315 | self.filename = getattr(filename, "name", "No name") |
| 2316 | |
| 2317 | def _write_header(self, pkt): |
| 2318 | # type: (Optional[Union[Packet, bytes]]) -> None |
| 2319 | self.header_present = True |
| 2320 | |
| 2321 | if self.append: |