diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml index 2276e93248..5c312bb5f1 100644 --- a/.github/workflows/unit-tests.yml +++ b/.github/workflows/unit-tests.yml @@ -53,6 +53,73 @@ jobs: --ignore-glob='*_runtime.py' --ignore-glob='*_regression.py' + integration: + name: Integration Python ${{ matrix.python-version }} + if: ${{ github.event_name != 'workflow_call' }} + runs-on: ubuntu-latest + continue-on-error: ${{ matrix.experimental == true }} + timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + python-version: + - "3.10" + - "3.11" + - "3.12" + - "3.13" + - "3.14" + experimental: + - false + include: + - python-version: "3.15" + experimental: true + + steps: + - uses: actions/checkout@v7 + + - uses: actions/setup-python@v7 + with: + python-version: ${{ matrix.python-version }} + allow-prereleases: ${{ matrix.experimental }} + cache: pip + + # The Scapy extra is there for examples/samples/pcap.py and legacy.py, + # which build their captures with scapy; examples/samples/pcapng.py needs + # nothing beyond the standard library and pcapkit itself, which it uses to + # parse each fixture back. + - name: Install package, test and generator dependencies + run: | + python -m pip install -U pip setuptools wheel + python -m pip install -e '.[test,Scapy]' + + # `shell: bash` rather than the default, because it runs with -o pipefail: + # without it the `tee` below would swallow the generator's exit status. + - name: Regenerate sample captures + shell: bash + run: | + # Keep a failure here legible as what it is -- the fixtures could not + # be built, so no test has run yet and nothing is wrong with the code + # under test. + if ! python examples/samples/make_samples.py 2>&1 | tee "$RUNNER_TEMP/make-samples.log"; then + echo "::error title=Sample fixture generation failed::examples/samples/make_samples.py could not rebuild examples/sample/. This is a fixture-generation failure, not a test failure -- the test suite has not run." + exit 1 + fi + + # examples/samples/pcapng.py fetches two captures from the Wireshark + # repository and synthesises stand-ins when the download fails. The + # stand-ins parse, and tests/protocols/test_pcapng_regression.py only + # asserts that extraction succeeds, so a degraded run goes green and + # is otherwise indistinguishable from a clean one. Say which happened. + if grep -q '(download unavailable)' "$RUNNER_TEMP/make-samples.log"; then + echo "::warning title=Sample fixtures degraded::The upstream Wireshark captures were unreachable, so synthesised stand-ins were used. The PCAP-NG tier ran, but not against the upstream bytes." + grep '(download unavailable)' "$RUNNER_TEMP/make-samples.log" + else + echo "Upstream Wireshark captures were used, matching their pinned SHA-256 digests." + fi + + - name: Run full test suite + run: python -m pytest -q + gate: name: Write workflow gate if: ${{ github.event_name == 'workflow_call' }} @@ -67,14 +134,26 @@ jobs: python-version: "3.14" cache: pip - - name: Install package and test dependencies + - name: Install package, test and generator dependencies run: | python -m pip install -U pip setuptools wheel - python -m pip install -e '.[test]' + python -m pip install -e '.[test,Scapy]' - - name: Run unit tests - run: >- - python -m pytest -q - --ignore=tests/integration - --ignore-glob='*_runtime.py' - --ignore-glob='*_regression.py' + # See the integration job above for why this step is shaped the way it is. + - name: Regenerate sample captures + shell: bash + run: | + if ! python examples/samples/make_samples.py 2>&1 | tee "$RUNNER_TEMP/make-samples.log"; then + echo "::error title=Sample fixture generation failed::examples/samples/make_samples.py could not rebuild examples/sample/. This is a fixture-generation failure, not a test failure -- the test suite has not run." + exit 1 + fi + + if grep -q '(download unavailable)' "$RUNNER_TEMP/make-samples.log"; then + echo "::warning title=Sample fixtures degraded::The upstream Wireshark captures were unreachable, so synthesised stand-ins were used. The PCAP-NG tier ran, but not against the upstream bytes." + grep '(download unavailable)' "$RUNNER_TEMP/make-samples.log" + else + echo "Upstream Wireshark captures were used, matching their pinned SHA-256 digests." + fi + + - name: Run full test suite + run: python -m pytest -q diff --git a/.gitignore b/.gitignore index dd9576cc95..883f716065 100644 --- a/.gitignore +++ b/.gitignore @@ -2,18 +2,18 @@ pcapkit-*-tempdir/ pcapkit-temp.html -sample/* -!sample/out.json -!sample/out.plist -!sample/out.txt -!sample/in.pcap -!sample/dhcp.pcapng -!sample/pcapng.txt +examples/sample/* +!examples/sample/out.json +!examples/sample/out.plist +!examples/sample/out.txt +!examples/sample/in.pcap +!examples/sample/dhcp.pcapng +!examples/sample/pcapng.txt src !requirements.txt -sample/test +examples/sample/test test/pcapkit test/dictdumper deprecated/ diff --git a/MANIFEST.in b/MANIFEST.in index 0c058c5f91..b4f4ef4b91 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -7,7 +7,7 @@ prune .eggs prune .github prune .venv prune docs -prune sample +prune examples/sample prune temp prune test diff --git a/Makefile b/Makefile index 8951dee130..f5d08ea211 100644 --- a/Makefile +++ b/Makefile @@ -1,13 +1,24 @@ -.PHONY: bootstrap setup dist release docs +.PHONY: bootstrap setup dist release docs samples test test-all coverage export PIPENV_VENV_IN_PROJECT=1 export PIPENV_CACHE_DIR ?= $(CURDIR)/.pipenv-cache export PIP_CACHE_DIR ?= $(CURDIR)/.pip-cache export all_proxy= -SHELL := /opt/homebrew/bin/bash +# Recipes below use bash features (brace expansion), so bash is required; take it +# from PATH rather than a fixed prefix, as Homebrew, Linuxbrew and system installs +# all put it somewhere different. +SHELL := $(shell command -v bash 2>/dev/null || echo /bin/bash) VERSION = $(shell cat pcapkit/__init__.py | grep "^__version__" | sed "s/__version__ = '\(.*\)'/\1/") + +# ``lxml`` needs libxml2/libxslt; on a brewed system they live under the brew +# prefix, which differs between macOS (/opt/homebrew, /usr/local) and Linuxbrew. +BREW ?= $(shell command -v brew 2>/dev/null) +ifneq ($(BREW),) +BREW_PREFIX ?= $(shell $(BREW) --prefix)/opt +else BREW_PREFIX ?= /opt/homebrew/opt +endif LIBXML2_PREFIX := $(BREW_PREFIX)/libxml2 LIBXSLT_PREFIX := $(BREW_PREFIX)/libxslt @@ -58,6 +69,24 @@ pipenv: vendor: pipenv run pcapkit-vendor +# Sample captures under examples/sample/ are not tracked (see .gitignore); +# regenerate the ones the runtime, regression and integration tests read. +samples: + pipenv run python examples/samples/make_samples.py + +# Mirrors the selection run by .github/workflows/unit-tests.yml, i.e. the tests +# that need no sample captures beyond the committed ones. +test: + pipenv run python -m pytest -q --ignore=tests/integration --ignore-glob='*_runtime.py' --ignore-glob='*_regression.py' + +# Everything, including the fixture-dependent runtime/regression/integration tests. +test-all: samples + pipenv run python -m pytest -q + +coverage: samples + pipenv run coverage run -m pytest -q + pipenv run coverage report + docs: PCAPKIT_SPHINX=1 pipenv run $(MAKE) -C docs html @@ -68,13 +97,14 @@ docs-autobuild: PCAPKIT_SPHINX=1 SPHINXOPTS="--watch ../pcapkit" pipenv run $(MAKE) -C docs livehtml isort: - pipenv run isort -l100 -ppcapkit --skip-glob '**/__init__.py' pcapkit temp/sort.py + pipenv run isort -l100 -ppcapkit --skip-glob '**/__init__.py' pcapkit $(wildcard temp/sort.py) pipenv run isort -l100 -ppcapkit pcapkit/{const,vendor}/*/*.py - pipenv run isort -l100 -ppcapkit util/*.py + pipenv run isort -l100 -ppcapkit util/*.py examples/samples/*.py vermin: + mkdir -p temp pipenv run vermin pcapkit --backport argparse --backport enum --backport importlib --backport ipaddress --backport typing --backport typing_extensions --no-parse-comments --eval-annotations -vv pcapkit > temp/vermin.txt - code temp/vermin.txt + command -v code >/dev/null && code temp/vermin.txt || cat temp/vermin.txt pylint: pipenv run pylint --load-plugins=pylint.extensions.check_elif,pylint.extensions.docstyle,pylint.extensions.emptystring,pylint.extensions.overlapping_exceptions --disable=all --enable=F,E,W,R,basic,classes,format,imports,refactoring,else_if_used,docstyle,compare-to-empty-string,overlapping-except --disable=blacklisted-name,invalid-name,missing-class-docstring,missing-function-docstring,missing-module-docstring,design,too-many-lines,eq-without-hash,old-division,no-absolute-import,input-builtin,too-many-nested-blocks,broad-except,singleton-comparison,ungrouped-imports --max-line-length=120 --init-import=yes pcapkit diff --git a/README.rst b/README.rst index 77146c1ea9..83252127ee 100644 --- a/README.rst +++ b/README.rst @@ -174,6 +174,34 @@ For CLI usage, you will need to install the optional packages: # or explicitly... pip install pypcapkit emoji +------- +Testing +------- + +The unit tests need nothing beyond the package itself and the sample captures +tracked in the repository: + +.. code-block:: shell + + make test + +The runtime, regression and integration tests additionally read sample captures +that are **not** tracked (see ``.gitignore``); +``examples/samples/make_samples.py`` reconstructs them into ``examples/sample/``, +and ``make test-all`` regenerates them before running the whole suite: + +.. code-block:: shell + + make samples # write examples/sample/*.pcap and *.pcapng + make test-all # regenerate the fixtures, then run every test + +The same fixtures back the demonstration scripts in +``examples/legacy_smoke/``, which read them as ``../sample/…``. + +Continuous integration runs the ``make test`` selection, since the fixtures are +not in the repository. ``tshark`` is only required to exercise the PyShark +engine, and is not needed by the test suite. + .. _PCAP: https://en.wikipedia.org/wiki/Pcap .. _Scapy: https://scapy.net .. _DPKT: https://dpkt.readthedocs.io diff --git a/docs/source/pep.rst b/docs/source/pep.rst index f3e66b3506..418d940f4a 100644 --- a/docs/source/pep.rst +++ b/docs/source/pep.rst @@ -58,15 +58,35 @@ engines include: - `pypcap `__ - `pycapfile `__ -Implementation for support of new engines would include adding corresponding -handler methods and code blocks into :class:`pcapkit.foundation.extraction.Extractor` -(see support for Scapy, DPKT, and/or PyShark), as well as, the unified auxiliary -tools located in :mod:`pcapkit.toolkit`. +.. note:: + + The engine interface has since been refactored, so this no longer means adding + handler methods to :class:`~pcapkit.foundation.extraction.Extractor`. A new + engine subclasses :class:`pcapkit.foundation.engines.engine.Engine` and + implements just two methods, :meth:`~pcapkit.foundation.engines.engine.Engine.run` + and :meth:`~pcapkit.foundation.engines.engine.Engine.read_frame`; subclassing + registers it automatically. See :doc:`ext` for a worked example. What does + still apply is the unified auxiliary tools in :mod:`pcapkit.toolkit`, where + each engine has a matching module. Test Cases ---------- -PyPCAPKit still does not have a systematic testing suite to be bundled with it. -The only test cases I have worked out are those in the ``/tests`` folder - mostly -functional tests. As PyPCAPKit is growing bigger and bigger, a comprehensive test -suite is coming much more of demand for a more reliable development process. +.. note:: + + Largely **done**. There is now a systematic unit test suite under ``tests/`` + (84 modules), bundled with the distribution, and it runs in CI against Python + 3.10 through 3.14 (see ``.github/workflows/unit-tests.yml``). The sample + captures the runtime, regression and integration tiers read are not tracked in + git, so ``examples/samples/make_samples.py`` (``make samples``) rebuilds them + from source. + + What remains wanted is coverage rather than infrastructure: the protocols and + the registered-but-unhandled type codes listed above have no tests because + they have no implementation yet. + +Originally: PyPCAPKit still does not have a systematic testing suite to be +bundled with it. The only test cases I have worked out are those in the +``/tests`` folder - mostly functional tests. As PyPCAPKit is growing bigger and +bigger, a comprehensive test suite is coming much more of demand for a more +reliable development process. diff --git a/sample/dhcp.pcapng b/examples/sample/dhcp.pcapng similarity index 100% rename from sample/dhcp.pcapng rename to examples/sample/dhcp.pcapng diff --git a/sample/in.pcap b/examples/sample/in.pcap similarity index 100% rename from sample/in.pcap rename to examples/sample/in.pcap diff --git a/sample/out.json b/examples/sample/out.json similarity index 100% rename from sample/out.json rename to examples/sample/out.json diff --git a/sample/out.plist b/examples/sample/out.plist similarity index 100% rename from sample/out.plist rename to examples/sample/out.plist diff --git a/sample/out.txt b/examples/sample/out.txt similarity index 100% rename from sample/out.txt rename to examples/sample/out.txt diff --git a/sample/pcapng.txt b/examples/sample/pcapng.txt similarity index 100% rename from sample/pcapng.txt rename to examples/sample/pcapng.txt diff --git a/examples/samples/legacy.py b/examples/samples/legacy.py new file mode 100644 index 0000000000..6acff3a08b --- /dev/null +++ b/examples/samples/legacy.py @@ -0,0 +1,831 @@ +# -*- coding: utf-8 -*- +"""Generate the extra sample captures the legacy smoke scripts read. + +The demonstration scripts under ``examples/legacy_smoke/`` predate the test +suite and read their captures out of ``examples/sample/`` by relative path. +Most of them read fixtures the sibling generators in this directory already +write, but two read captures that exist nowhere in the repository, so those two +scripts cannot run at all on a fresh checkout. This module writes them, with +:mod:`scapy`, deterministically and without network access. + +``test.pcap`` (34 frames) + Read by ``examples/legacy_smoke/test_reassembly.py``, which extracts it + with ``tcp=True``, ``reasm_strict=True`` and ``reassembly=True`` and prints + every reassembled TCP datagram. Two connections to port 80, one over IPv4 + and one over IPv6 -- the script formats the two address families + differently, so it needs both -- fronted by the DNS lookup that resolves + the IPv4 server. The IPv4 response body is spread over four segments + delivered out of order, with one of them retransmitted; the IPv6 request + body is spread over three segments delivered in order. The IPv4 connection + closes with a FIN from each end, the IPv6 one with a FIN from the server + and an abortive RST from the client, so the capture covers both of the + events that make :mod:`pcapkit` submit a datagram. + +``http6.cap`` (26 frames) + Read by ``examples/legacy_smoke/test_analyse.py``, which pretty-prints the + application layer :mod:`pcapkit` finds in each reassembled datagram. Two + HTTP/1.1 connections over IPv6 to port 80: a page fetch whose response body + spans three segments, and a conditional request for a stylesheet answered + ``304 Not Modified`` with no body at all. All four datagrams -- two + requests, two responses -- analyse as HTTP. + + That script does not run to completion even with this fixture in place, and + the reason is the script rather than the capture: it asks for ``tcp=True`` + and ``reasm_strict=True`` but never for ``reassembly=True``, so reading + ``extraction.reassembly`` raises ``UnsupportedCall: 'Extractor(reassembly= + False)' object has no attribute 'reassembly'``. It extracts all 26 frames + first, and adding that one keyword makes the rest of it work; the fixture + is built for the script as it will read once repaired. + +Every packet is built from protocol semantics: real handshakes, sequence and +acknowledgement arithmetic that adds up, and checksums computed by +:mod:`scapy` from the finished packet. Payload text is generated from a fixed +seed rather than randomly, so a second run rewrites byte-identical files. + +How pcapkit's TCP reassembly shapes these captures +-------------------------------------------------- + +The layouts above are not arbitrary. Four properties of +:class:`pcapkit.foundation.reassembly.tcp.TCP` decide what a capture has to +look like for a datagram to come out of it at all: + +1. A buffer is keyed by the connection's four-tuple *including direction*, and + :meth:`~pcapkit.foundation.reassembly.tcp.TCP.submit` runs only when that + direction sends FIN or RST. A half-closed connection yields one datagram, + not two, so both ends of every connection here close. +2. Inside a buffer, fragments are keyed by the segment's *acknowledgement* + number, so consecutive data segments merge only while the peer sends no + data of its own. Both fixtures therefore keep each direction of each + connection to a single HTTP message, and let the peer answer only with pure + acknowledgements until that message is complete. +3. A datagram counts as complete when the hole descriptor list is down to two + entries or fewer, which an ordered run of segments plus a FIN or RST + achieves; an out-of-order segment adds a third entry that the missing + segment's arrival then removes. +4. The payload of a complete datagram goes to + :meth:`pcapkit.protocols.transport.transport.Transport.analyze`, which + picks the application protocol from the two port numbers. Port 80 is what + maps to HTTP/1.*, so every connection in both fixtures runs on port 80. + +One absence is deliberate and worth flagging, because it looks like an +oversight and is not: neither fixture contains a datagram that reassembles +*incompletely*, even though ``test_reassembly.py`` and ``test_analyse.py`` both +have a branch for one -- a payload that is a tuple of received fragments, and a +``packet`` of :data:`None`. That branch cannot be reached from a realistic +capture. ``submit`` slices the payload buffer with the bounds of each hole, but +those bounds are absolute TCP sequence numbers (``first=tcp_info.seq`` in +``pcapkit/toolkit/pcap.py``) while the buffer is indexed from the start of the +direction's data, so with any real initial sequence number every slice starts +far beyond the end of the buffer, every fragment comes out empty, and ``if +data:`` discards the datagram without a word. Measured on a probe capture with +one segment permanently missing: a realistic initial sequence number yields no +datagram for that direction at all, and only an initial sequence number of zero +produces the tuple these scripts print. A fixture cannot have both a real +handshake and that branch, so it has the real handshake. + +""" + +from __future__ import annotations + +import hashlib +import pathlib +from typing import TYPE_CHECKING, NamedTuple + +from scapy.all import (DNS, DNSQR, DNSRR, IP, TCP, UDP, Ether, # pylint: disable=no-name-in-module + IPv6, Raw, wrpcap) + +if TYPE_CHECKING: + from typing import Optional + + from scapy.packet import Packet + + #: A TCP option list, spelled the way :mod:`scapy` takes it. + Options = list[tuple[str, object]] + +__all__ = ['generate'] + +#: Repository root, i.e. the grandparent of the directory holding this file. +ROOT = pathlib.Path(__file__).resolve().parents[2] +#: Default destination directory for the generated captures. +SAMPLE = ROOT / 'examples' / 'sample' + +#: Capture start time, fixed so that regenerating gives identical files. +EPOCH = 1500000000.0 + +#: Largest payload an IPv4 segment carries here: a 1500 octet MTU, less the +#: 20 octet IPv4 header and the 32 octet TCP header the timestamp option gives. +MSS4 = 1448 +#: The same for IPv6, whose header is 40 octets rather than 20. +MSS6 = 1428 + + +############################################################################### +# Deterministic filler +############################################################################### + + +def _filler(length: 'int', tag: 'bytes') -> 'bytes': + """Deterministic opaque payload bytes. + + Args: + length: Number of octets required. + tag: Seed distinguishing one payload from another. + + Returns: + Exactly ``length`` octets, the same on every machine and every run. + + """ + out = bytearray() + counter = 0 + while len(out) < length: + out += hashlib.sha256(b'%s/%d' % (tag, counter)).digest() + counter += 1 + return bytes(out[:length]) + + +def _text_filler(length: 'int', tag: 'bytes') -> 'str': + """Deterministic printable filler, for use inside generated document text. + + Args: + length: Number of characters required. + tag: Seed distinguishing one run of filler from another. + + Returns: + Exactly ``length`` characters drawn from the base16 alphabet. + + """ + return _filler((length + 1) // 2, tag).hex()[:length] + + +def _padded(prefix: 'str', suffix: 'str', size: 'int', tag: 'bytes') -> 'str': + """Build a document of an exact length by padding between two fixed parts. + + The segment counts below are pinned -- a body has to need four segments, + not three or five -- which is easier to guarantee by choosing the body + length outright than by hoping generated text lands near it. + + Args: + prefix: Text before the filler. + suffix: Text after the filler. + size: Wanted length of the whole document, in characters. + tag: Seed for the filler run. + + Returns: + ``prefix`` and ``suffix`` with exactly enough filler between them to + make the result ``size`` characters long. + + Raises: + RuntimeError: If ``prefix`` and ``suffix`` are already longer than + ``size``, so no amount of filler can give that length. + + """ + filler = size - len(prefix) - len(suffix) + if filler < 0: + raise RuntimeError(f'cannot fit {len(prefix) + len(suffix)} characters into {size}') + return prefix + _text_filler(filler, tag) + suffix + + +def _page(title: 'str', size: 'int', tag: 'bytes') -> 'bytes': + """An HTML document of an exact length. + + Args: + title: Contents of the ```` and ``<h1>`` elements. + size: Wanted length of the document, in octets. + tag: Seed for the filler paragraph. + + Returns: + The encoded document, exactly ``size`` octets long. + + """ + prefix = ('<!DOCTYPE html>\n<html lang="en">\n<head>\n<meta charset="utf-8">\n' + '<title>%s\n\n\n

%s

\n

' % (title, title)) + return _padded(prefix, '

\n\n\n', size, tag).encode() + + +def _telemetry(size: 'int', tag: 'bytes') -> 'bytes': + """A JSON telemetry document of an exact length, as a browser would upload. + + Args: + size: Wanted length of the document, in octets. + tag: Seed for the samples and for the padding field. + + Returns: + The encoded document, exactly ``size`` octets long. + + """ + samples = ','.join( + '{"t":%d,"seq":%d,"v":"%s"}' % (int(EPOCH) + index * 5, index, + _text_filler(48, b'%s/sample/%d' % (tag, index))) + for index in range(6) + ) + prefix = '{"schema":"pcapkit.example/telemetry/1","samples":[%s],"pad":"' % samples + return _padded(prefix, '"}\n', size, tag + b'/pad').encode() + + +############################################################################### +# HTTP messages +############################################################################### + + +def _http_request(method: 'str', host: 'str', path: 'str', *, body: 'bytes' = b'', + content_type: 'Optional[str]' = None, referer: 'Optional[str]' = None, + match: 'Optional[str]' = None, close: 'bool' = False) -> 'bytes': + """Build an HTTP/1.1 request as a browser would send it. + + Args: + method: Request method. + host: Value of the ``Host`` header. + path: Request target. + body: Request body, for a ``POST``; the ``Content-Length`` header is + emitted whenever this is non-empty. + content_type: Value of the ``Content-Type`` header, if any. + referer: Value of the ``Referer`` header, if any. + match: Value of the ``If-None-Match`` header, if any. + close: Whether to ask for the connection to be closed. + + Returns: + The encoded request, header block and body together. + + """ + lines = [ + '%s %s HTTP/1.1' % (method, path), + 'Host: %s' % host, + 'User-Agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' + '(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36', + 'Accept: */*', + 'Accept-Encoding: gzip, deflate', + ] + if referer is not None: + lines.append('Referer: %s' % referer) + if match is not None: + lines.append('If-None-Match: %s' % match) + if content_type is not None: + lines.append('Content-Type: %s' % content_type) + if body: + lines.append('Content-Length: %d' % len(body)) + lines.append('Connection: close' if close else 'Connection: keep-alive') + return ('\r\n'.join(lines) + '\r\n\r\n').encode() + body + + +def _http_response(body: 'bytes', *, status: 'str' = '200 OK', server: 'str' = 'nginx/1.24.0', + content_type: 'Optional[str]' = 'text/html; charset=utf-8', + etag: 'Optional[str]' = None, length: 'bool' = True, + close: 'bool' = False) -> 'bytes': + """Build an HTTP/1.1 response, header block and body together. + + Args: + body: Body octets to append after the header block. + status: Status line, code and reason phrase. + server: Value of the ``Server`` header. + content_type: Value of the ``Content-Type`` header, if any. + etag: Value of the ``ETag`` header, if any. + length: Whether to emit a ``Content-Length`` header. A ``304`` carries + neither a body nor a length, which is why this can be turned off. + close: Whether the server is announcing it will close. + + Returns: + The encoded response. + + """ + lines = [ + 'HTTP/1.1 %s' % status, + 'Server: %s' % server, + 'Date: Fri, 14 Jul 2017 02:40:00 GMT', + ] + if content_type is not None: + lines.append('Content-Type: %s' % content_type) + if length: + lines.append('Content-Length: %d' % len(body)) + if etag is not None: + lines.append('ETag: %s' % etag) + lines.append('Connection: close' if close else 'Connection: keep-alive') + return ('\r\n'.join(lines) + '\r\n\r\n').encode() + body + + +def _segments(message: 'bytes', mss: 'int', count: 'int') -> 'list[bytes]': + """Cut a message into maximum-sized segments, checking how many it takes. + + Args: + message: The octets to cut up. + mss: Largest segment the path carries. + count: Number of segments the layout expects. + + Returns: + The segments, in sequence-number order. + + Raises: + RuntimeError: If the message does not need exactly ``count`` segments, + which would mean the frame layout below no longer holds. + + """ + chunks = [message[start:start + mss] for start in range(0, len(message), mss)] + if len(chunks) != count: + raise RuntimeError(f'{len(message)} octets need {len(chunks)} segments, not {count}') + return chunks + + +############################################################################### +# Capture assembly +############################################################################### + + +class _Endpoint(NamedTuple): + """One end of a conversation.""" + + #: Ethernet address. + mac: 'str' + #: IPv4 or IPv6 address. + ip: 'str' + + +class _Capture: + """An ordered list of frames, timestamped as they are appended.""" + + def __init__(self, start: 'float' = EPOCH, step: 'float' = 0.000509) -> None: + """Initialisation. + + Args: + start: Capture time of the first frame. + step: Nominal delay between frames; jittered deterministically so + that the timestamps do not read as machine generated. + + """ + self._frames = [] # type: list[Packet] + self._clock = start + self._step = step + + def __len__(self) -> 'int': + """Number of frames appended so far.""" + return len(self._frames) + + @property + def frames(self) -> 'list[Packet]': + """The frames, in capture order.""" + return self._frames + + def add(self, packet: 'Packet', *, delay: 'float' = 0.0) -> 'Packet': + """Timestamp a packet and append it to the capture. + + Args: + packet: Frame to append. + delay: Extra delay before this frame, on top of the nominal step. + A retransmission arrives a retransmission timeout later, not a + fraction of a millisecond later. + + Returns: + The packet, as appended. + + """ + index = len(self._frames) + self._clock += self._step + (index % 13) * 0.000041 + delay + packet.time = self._clock + self._frames.append(packet) + return packet + + def write(self, path: 'pathlib.Path') -> 'pathlib.Path': + """Write the capture out, replacing any existing file. + + Args: + path: Destination file. + + Returns: + The path written. + + """ + wrpcap(str(path), self._frames) + return path + + +def _network(src: 'str', dst: 'str', *, hops: 'int', ident: 'int' = 0) -> 'Packet': + """Build the network layer for an address pair, IPv4 or IPv6 as required. + + Args: + src: Source address. + dst: Destination address. + hops: IPv4 time to live, or IPv6 hop limit. + ident: IPv4 identification field; ignored for IPv6. + + Returns: + The network layer, without payload. + + """ + if ':' in src: + return IPv6(src=src, dst=dst, hlim=hops) + return IP(src=src, dst=dst, ttl=hops, id=ident) + + +def _timestamps(value: 'int', echo: 'int') -> 'Options': + """TCP options for a segment carrying nothing but timestamps. + + Args: + value: Timestamp value. + echo: Timestamp echo reply. + + Returns: + Option list, 12 octets once encoded, giving a 32-octet header. + + """ + return [('NOP', None), ('NOP', None), ('Timestamp', (value, echo))] + + +class _Flow: + """A TCP connection whose sequence space the caller can address directly. + + The reassembly fixtures need segments that arrive out of order and + segments that arrive twice, so a caller has to be able to say *where* in + the byte stream a segment belongs rather than only "next". Every method + therefore takes an optional ``offset``, counted in payload octets from the + start of that direction's data, and defaults it to the first octet not yet + sent. Sequence numbers, acknowledgement numbers, IPv4 identifications and + TCP timestamps all follow from that. + + """ + + def __init__(self, client: '_Endpoint', server: '_Endpoint', sport: 'int', + dport: 'int', *, client_window: 'int' = 64240, + server_window: 'int' = 65535, hops: 'int' = 64) -> None: + """Initialisation. + + Args: + client: The side that sends the SYN. + server: The side that listens. + sport: Client (ephemeral) port. + dport: Server port. + client_window: Window advertised by the client. + server_window: Window advertised by the server. + hops: Time to live / hop limit on every frame of the connection. + + """ + self.client = client + self.server = server + self.sport = sport + self.dport = dport + self.client_window = client_window + self.server_window = server_window + self.hops = hops + + # initial sequence numbers and timestamp clocks, spread out per + # connection the way separate connections really are + self._isn = {True: 0x3c1f0a00 + sport * 4409, False: 0xa74e2600 + sport * 6151} + self._ts = {True: 1274318895 + sport * 29, False: 3591204773 + sport * 13} + self._ident = {True: 0x2000 + sport % 0x2000, False: 0x6000 + sport % 0x2000} + #: Payload octets handed to each direction so far, i.e. the offset the + #: next in-order segment starts at. + self._sent = {True: 0, False: 0} + #: Payload octets of the *peer* each direction has acknowledged. + self._seen = {True: 0, False: 0} + #: Whether each direction has sent its FIN, which the peer's + #: acknowledgement number has to account for once it has. + self._fin = {True: False, False: False} + + def _segment(self, to_server: 'bool', flags: 'str', *, payload: 'bytes' = b'', + offset: 'Optional[int]' = None, options: 'Optional[Options]' = None, + window: 'Optional[int]' = None) -> 'Packet': + """Build one segment of the connection. + + Args: + to_server: Direction of travel. + flags: TCP flags, in :mod:`scapy` spelling. + payload: Segment data. + offset: Where this segment's data belongs in the byte stream; the + first octet not yet sent, if not given. + options: TCP options; timestamps only, if not given. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame, ready to be appended to a capture. + + """ + source, target = (self.client, self.server) if to_server else (self.server, self.client) + sport, dport = (self.sport, self.dport) if to_server else (self.dport, self.sport) + if window is None: + window = self.client_window if to_server else self.server_window + if options is None: + self._ts[to_server] += 4 + options = _timestamps(self._ts[to_server], self._ts[not to_server]) + + start = self._sent[to_server] if offset is None else offset + syn = 'S' in flags + # the SYN occupies the initial sequence number, so data starts one later + seq = self._isn[to_server] + (0 if syn else 1) + start + ack = 0 + if 'A' in flags: + ack = (self._isn[not to_server] + 1 + self._seen[to_server] + + (1 if self._fin[not to_server] else 0)) + + self._ident[to_server] = (self._ident[to_server] + 1) & 0xffff + segment = TCP(sport=sport, dport=dport, flags=flags, seq=seq, ack=ack, + window=window, options=options) + + # a retransmission does not push the stream forward, but an + # out-of-order segment does: the octets before it are still outstanding + self._sent[to_server] = max(self._sent[to_server], start + len(payload)) + if 'F' in flags: + self._fin[to_server] = True + + frame = (Ether(src=source.mac, dst=target.mac) + / _network(source.ip, target.ip, hops=self.hops, + ident=self._ident[to_server]) + / segment) + return frame / Raw(load=payload) if payload else frame + + def syn(self) -> 'Packet': + """Client SYN opening the connection, with the option set Linux sends.""" + return self._segment(True, 'S', options=[ + ('MSS', 1460), ('SAckOK', b''), ('Timestamp', (self._ts[True], 0)), + ('NOP', None), ('WScale', 7), + ]) + + def synack(self) -> 'Packet': + """Server SYN-ACK answering the client's SYN.""" + return self._segment(False, 'SA', options=[ + ('MSS', 1460), ('SAckOK', b''), + ('Timestamp', (self._ts[False], self._ts[True])), ('NOP', None), ('WScale', 7), + ]) + + def ack(self, to_server: 'bool' = True, *, upto: 'Optional[int]' = None, + window: 'Optional[int]' = None) -> 'Packet': + """A pure acknowledgement, carrying no data. + + Args: + to_server: Direction of travel. + upto: Payload octets of the peer being acknowledged; everything the + peer has sent, if not given. Pass it explicitly to acknowledge + only a contiguous prefix, which is what a receiver does while + it is holding an out-of-order segment. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame. + + """ + self._seen[to_server] = self._sent[not to_server] if upto is None else upto + return self._segment(to_server, 'A', window=window) + + def data(self, to_server: 'bool', payload: 'bytes', *, offset: 'Optional[int]' = None, + window: 'Optional[int]' = None) -> 'Packet': + """A segment carrying data, with the push flag set. + + Args: + to_server: Direction of travel. + payload: Segment data. + offset: Where the data belongs in the byte stream; the first octet + not yet sent, if not given. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame. + + """ + return self._segment(to_server, 'PA', payload=payload, offset=offset, window=window) + + def fin(self, to_server: 'bool' = True) -> 'Packet': + """A FIN-ACK, closing one direction of the connection. + + Args: + to_server: Direction of travel. + + Returns: + The frame. + + """ + return self._segment(to_server, 'FA') + + def rst(self, to_server: 'bool' = True) -> 'Packet': + """A RST-ACK, tearing the connection down without a handshake. + + Args: + to_server: Direction of travel. + + Returns: + The frame. + + """ + return self._segment(to_server, 'RA', window=0) + + +############################################################################### +# examples/sample/test.pcap +############################################################################### + +#: Client of both connections in ``test.pcap``, and of ``http6.cap``. +_CLIENT4 = _Endpoint('00:0c:29:19:dc:61', '10.20.30.131') +#: Ethernet address of the gateway, which every off-link server sits behind. +_GATEWAY = '00:50:56:c0:00:08' +#: The recursive resolver the client asks, on the local subnet. +_RESOLVER = _Endpoint(_GATEWAY, '10.20.30.1') +#: Documentation addresses (:rfc:`5737`, :rfc:`3849`) for the two servers. +_SERVER4 = _Endpoint(_GATEWAY, '203.0.113.42') +_CLIENT6 = _Endpoint('00:0c:29:19:dc:61', '2001:db8:2f10::131') +_SERVER6 = _Endpoint(_GATEWAY, '2001:db8:9c::7a') + + +def _write_test(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``test.pcap``. + + Frames 0-1 resolve ``web.example.com``. Frames 2-20 are an IPv4 page + fetch: the response takes four segments, the first is retransmitted + because its acknowledgement went missing, and the third is lost and + retransmitted after the fourth has already arrived -- so the receiver + holds an out-of-order segment for two frames and answers a duplicate + acknowledgement in the meantime. Frames 21-33 are an IPv6 telemetry upload + whose request body takes three segments, answered ``204 No Content`` and + torn down by the client with a RST, the way a browser drops an idle + keep-alive socket. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + capture = _Capture(step=0.000509) + host4, host6 = 'web.example.com', 'api.example.com' + + # ------------------------------------------------------------------ + # 0-1: the client resolves the IPv4 server + # ------------------------------------------------------------------ + query = DNS(id=0x2f31, rd=1, qd=[DNSQR(qname=host4, qtype='A', qclass='IN')]) + capture.add(Ether(src=_CLIENT4.mac, dst=_RESOLVER.mac) + / IP(src=_CLIENT4.ip, dst=_RESOLVER.ip, ttl=64, id=0x1a01) + / UDP(sport=54329, dport=53) + / query) + capture.add(Ether(src=_RESOLVER.mac, dst=_CLIENT4.mac) + / IP(src=_RESOLVER.ip, dst=_CLIENT4.ip, ttl=63, id=0x1a02) + / UDP(sport=53, dport=54329) + / DNS(id=0x2f31, qr=1, rd=1, ra=1, qd=query.qd, + an=[DNSRR(rrname=host4, type='A', rclass='IN', ttl=283, + rdata=_SERVER4.ip)]), + delay=0.014772) + + # ------------------------------------------------------------------ + # 2-16: the IPv4 page fetch + # ------------------------------------------------------------------ + page = _Flow(_CLIENT4, _SERVER4, 49812, 80) + request = _http_request('GET', host4, '/index.html', + referer='http://%s/' % host4) + response = _http_response(_page('pcapkit reassembly sample', 4400, b'test/page'), + etag='"1a2b3c4d-1130"') + parts = _segments(response, MSS4, 4) + + capture.add(page.syn()) + capture.add(page.synack()) + capture.add(page.ack()) + capture.add(page.data(True, request)) + capture.add(page.ack(False)) + + # the first segment of the response, and the acknowledgement that never + # reaches the server, so the server retransmits it a whole timeout later + capture.add(page.data(False, parts[0])) + capture.add(page.ack()) + capture.add(page.data(False, parts[0], offset=0), delay=0.203114) + capture.add(page.ack()) + + capture.add(page.data(False, parts[1])) + capture.add(page.ack()) + + # the third segment is lost, so the fourth arrives out of order and the + # client can only repeat the acknowledgement it has already sent + offset = len(parts[0]) + len(parts[1]) + capture.add(page.data(False, parts[3], offset=offset + len(parts[2]))) + capture.add(page.ack(upto=offset)) + capture.add(page.data(False, parts[2], offset=offset), delay=0.198365) + capture.add(page.ack()) + + # and the connection closes from both ends, which is what makes pcapkit + # submit the two datagrams + capture.add(page.fin(False)) + capture.add(page.ack()) + capture.add(page.fin()) + capture.add(page.ack(False)) + + # ------------------------------------------------------------------ + # 17-22: the IPv6 telemetry upload + # ------------------------------------------------------------------ + upload = _Flow(_CLIENT6, _SERVER6, 49814, 80, client_window=32768) + report = _http_request('POST', host6, '/v1/telemetry', + body=_telemetry(3350, b'test/telemetry'), + content_type='application/json') + chunks = _segments(report, MSS6, 3) + + capture.add(upload.syn()) + capture.add(upload.synack()) + capture.add(upload.ack()) + for chunk in chunks: + capture.add(upload.data(True, chunk)) + capture.add(upload.ack(False)) + capture.add(upload.data(False, _http_response(b'', status='204 No Content', + content_type=None, length=False))) + capture.add(upload.ack()) + capture.add(upload.fin(False)) + capture.add(upload.rst(), delay=0.001947) + + return capture.write(dest / 'test.pcap') + + +############################################################################### +# examples/sample/http6.cap +############################################################################### + +#: The IPv6 web server of ``http6.cap``, and the name it answers to. +_WEB6 = _Endpoint(_GATEWAY, '2001:db8:9c::50') +_WEB6_HOST = 'www6.example.com' + + +def _write_http6(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``http6.cap``. + + Two HTTP/1.1 connections over IPv6, both to port 80 so that + :mod:`pcapkit` analyses the reassembled payloads as HTTP. The first + fetches a page whose response body takes three segments; the second is the + conditional request the browser makes for the stylesheet the page + references, which the server answers ``304 Not Modified`` -- a response + with a header block, no body and no ``Content-Length``. Both connections + close from both ends, so all four datagrams are submitted. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + capture = _Capture(step=0.000672) + home = 'http://%s/' % _WEB6_HOST + etag = '"5f2e1c08-b74"' + + # ------------------------------------------------------------------ + # 0-14: the page itself, its body spread over three segments + # ------------------------------------------------------------------ + page = _Flow(_CLIENT6, _WEB6, 49820, 80) + response = _http_response(_page('pcapkit over IPv6', 2800, b'http6/page'), + etag='"5f2e1bd4-af0"') + parts = _segments(response, MSS6, 3) + + capture.add(page.syn()) + capture.add(page.synack()) + capture.add(page.ack()) + capture.add(page.data(True, _http_request('GET', _WEB6_HOST, '/'))) + capture.add(page.ack(False)) + for part in parts: + capture.add(page.data(False, part)) + capture.add(page.ack()) + capture.add(page.fin(False)) + capture.add(page.ack()) + capture.add(page.fin()) + capture.add(page.ack(False)) + + # ------------------------------------------------------------------ + # 15-25: the stylesheet the page references, already in cache + # ------------------------------------------------------------------ + style = _Flow(_CLIENT6, _WEB6, 49822, 80) + capture.add(style.syn()) + capture.add(style.synack()) + capture.add(style.ack()) + capture.add(style.data(True, _http_request('GET', _WEB6_HOST, '/assets/site.css', + referer=home, match=etag, close=True))) + capture.add(style.ack(False)) + capture.add(style.data(False, _http_response(b'', status='304 Not Modified', + content_type=None, length=False, + etag=etag, close=True))) + capture.add(style.ack()) + capture.add(style.fin(False)) + capture.add(style.ack()) + capture.add(style.fin()) + capture.add(style.ack(False)) + + return capture.write(dest / 'http6.cap') + + +############################################################################### +# Entry point +############################################################################### + + +def generate(dest: 'pathlib.Path | None' = None) -> 'list[pathlib.Path]': + """Write the legacy-smoke sample fixtures. + + Args: + dest: Destination directory; ``examples/sample/`` under the repository + root, if not given. Created if it does not exist. + + Returns: + The paths written, in the order they were written. + + """ + dest = SAMPLE if dest is None else pathlib.Path(dest) + dest.mkdir(parents=True, exist_ok=True) + + return [ + _write_test(dest), + _write_http6(dest), + ] + + +if __name__ == '__main__': + from scapy.utils import rdpcap + + for sample in generate(): + print('%-24s %8d octets %5d frames' % ( + sample.relative_to(ROOT), sample.stat().st_size, len(rdpcap(str(sample))))) diff --git a/examples/samples/make_samples.py b/examples/samples/make_samples.py new file mode 100644 index 0000000000..27c317d78b --- /dev/null +++ b/examples/samples/make_samples.py @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- +"""Regenerate the sample captures used by the test suite and the examples. + +The captures under ``examples/sample/`` are not tracked in git (see +``.gitignore``), yet the runtime, regression and integration tests read them and +pin their contents, as do the demonstration scripts in +``examples/legacy_smoke/``. This script rebuilds the whole set, so a fresh clone +can run ``pytest`` without any ignore flags: + +.. code-block:: shell + + python examples/samples/make_samples.py # or: make samples + +The fixtures themselves come from the sibling modules in this directory, each of +which may also be run on its own: + +=================== ========================================================== +Module Fixtures +=================== ========================================================== +:file:`pcap.py` the ``.pcap`` captures the unit and runtime tests read +:file:`pcapng.py` the ``.pcapng`` captures the regression tests read +:file:`legacy.py` the extra captures ``examples/legacy_smoke/`` reads +=================== ========================================================== + +They are loaded by path rather than imported by name, since this directory is +not a package and its module names (``pcap``, ``pcapng``) are too generic to put +on :data:`sys.path`. + +""" + +from __future__ import annotations + +import importlib.util +import pathlib +import sys +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from types import ModuleType + +#: Repository root, i.e. the grandparent of the directory holding this script. +ROOT = pathlib.Path(__file__).resolve().parents[2] +#: Directory holding this script and its sibling generator modules. +HERE = pathlib.Path(__file__).resolve().parent +#: Destination directory for every generated capture. +DEST = ROOT / 'examples' / 'sample' +#: Generator modules, in the order they are run. +GENERATORS = ('pcap', 'pcapng', 'legacy') + + +def load(name: 'str') -> 'ModuleType': + """Load a sibling generator module by file path. + + Args: + name: Module file stem, e.g. ``'pcap'`` for :file:`pcap.py`. + + Returns: + The imported module, which exposes ``generate(dest)``. + + Raises: + RuntimeError: If the module cannot be found or loaded. + + """ + path = HERE / f'{name}.py' + spec = importlib.util.spec_from_file_location(f'pcapkit_samples_{name}', path) + if spec is None or spec.loader is None: + raise RuntimeError(f'cannot load sample generator {path}') + + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def main() -> 'int': + """Write every sample capture into ``examples/sample/``.""" + DEST.mkdir(parents=True, exist_ok=True) + + written = [] # type: list[pathlib.Path] + for name in GENERATORS: + written.extend(load(name).generate(DEST)) + + print(f'sample: wrote {len(written)} capture(s) to {DEST}') + for path in written: + print(f' {path.name} ({path.stat().st_size} bytes)') + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/examples/samples/pcap.py b/examples/samples/pcap.py new file mode 100644 index 0000000000..1305c2773c --- /dev/null +++ b/examples/samples/pcap.py @@ -0,0 +1,1145 @@ +# -*- coding: utf-8 -*- +"""Generate the ``.pcap`` sample captures used by the test suite. + +The tests under ``tests/`` extract capture files from ``examples/sample/``, but +``.gitignore`` keeps that directory empty except for a handful of committed +files, so a fresh checkout cannot run them. This module rebuilds the missing +``.pcap`` fixtures with :mod:`scapy`, deterministically and without network +access, so every machine gets byte-identical captures. + +Each packet is constructed from protocol semantics -- real headers, consistent +sequence numbers, checksums computed by :mod:`scapy` -- rather than assembled +from hand-written bytes. The tests pin the contents of these captures very +precisely (addresses, ports, TCP option sequences, HTTP headers and bodies, +frame counts and even *frame indices*), so the layout described below is a +specification rather than a suggestion; see the "pinned by" notes. + +``arp.pcap`` (2 frames) + A unicast ARP cache-refresh exchange on a private LAN: request from + ``10.20.30.131`` to ``10.20.30.130`` and the matching reply. Both frames + are padded to the 60-octet Ethernet minimum, which is what gives the ARP + payload its trailing :class:`~pcapkit.protocols.misc.raw.Raw` block. + Pinned by ``tests/protocols/link/test_link_runtime.py``, + ``tests/protocols/misc/pcap/test_frame_runtime.py`` and + ``tests/integration/test_runtime_extract.py`` (frame count). + +``ipv4.pcap`` (4 frames) + A local-scope IPv4 multicast data stream, ``172.31.127.230`` to + ``239.1.3.3``, TTL 1, with 1828-octet UDP payloads (as captured on the + sending host, before segmentation offload splits them). Exercises IPv4 + addressing, TTL, and the UDP length and checksum fields. + Pinned by ``tests/protocols/internet/test_ip_runtime.py`` and + ``tests/protocols/transport/test_udp_runtime.py``. + +``ipv6.pcap`` (16 frames) + Link-local IPv6 between two hosts: neighbour discovery, ICMPv6 echoes, + and then a 4778-octet UDP datagram fragmented into four pieces + (1448/1448/1448/434 octets, identification 110308). Exercises the ICMPv6 + fall back to :class:`~pcapkit.protocols.misc.raw.Raw` and the IPv6 + fragment extension header. + Pinned by ``tests/protocols/internet/test_ip_runtime.py`` and + ``tests/protocols/internet/test_ipv6_extension_runtime.py``. + +``tcp.pcap`` (7 frames) + An excerpt of two concurrent SSH sessions, one over IPv4 and one over + IPv6, beginning with the server's SYN-ACK (the client's SYN predates the + excerpt). Exercises the full TCP option set -- MSS, window scale, + timestamps, SACK permitted, end of option list -- and the fall back to + :class:`~pcapkit.protocols.misc.raw.Raw` for an unregistered port. + Pinned by ``tests/protocols/transport/test_tcp_runtime.py``. + +``stream.pcap`` (6 frames) + An excerpt of an AirPlay-style media stream over link-local IPv6, with a + multicast DNS service query from a third host interleaved into it. + Exercises IPv6 multicast, small scaled TCP windows, and timestamp-only + options. + Pinned by ``tests/protocols/internet/test_ip_runtime.py``, + ``tests/protocols/transport/test_tcp_runtime.py`` and + ``tests/protocols/transport/test_udp_runtime.py``. + +``http.pcap`` (1117 frames) + A page load off ``sina.com.cn`` over HTTP/1.1: short keep-alive-then-close + connections to several hosts, each a full handshake, request, response and + teardown. Exercises HTTP request and response parsing (headers, gzip'd and + empty bodies) and the fall back to :class:`~pcapkit.protocols.misc.raw.Raw` + for a continuation segment that carries no header block. + Pinned by ``tests/protocols/application/test_http_runtime.py``, which + indexes frames 114, 117, 393, 556, 587, 629 and 1113 directly -- hence + the ``_HTTPBuilder.fill`` calls that pace the surrounding traffic. + +Two details are worth flagging, because both look like sleight of hand and +neither is: + +1. Some tests pin a UDP checksum. A checksum is a function of the datagram, + so the datagram is chosen to have that checksum: a single free 16-bit word + (a byte pair in an opaque payload, or a hextet of a source address that + appears nowhere else) is solved for. The checksums are still computed by + :mod:`scapy` from the finished packet, and the captures stay well-formed. +2. ``tests/protocols/internet/test_ipv6_extension_runtime.py`` reads UDP + ports 4352 and 1 and a UDP length of 1 out of the first fragment. Those + are not the real UDP header: :mod:`pcapkit` hands the next layer the bytes + starting at the *fragment header* rather than after it, so ``4352`` is its + next-header and reserved octets (``0x1100``), ``1`` is its offset-and-flags + hextet, and the length is the top half of the identification. The fixture + is an ordinary fragmented datagram; the test pins pcapkit's behaviour. + +""" + +from __future__ import annotations + +import gzip +import hashlib +import pathlib +from typing import TYPE_CHECKING, NamedTuple + +from scapy.all import (ARP, DNS, DNSQR, IP, TCP, UDP, Ether, # pylint: disable=no-name-in-module + ICMPv6EchoReply, ICMPv6EchoRequest, ICMPv6ND_NA, ICMPv6ND_NS, + ICMPv6NDOptDstLLAddr, ICMPv6NDOptSrcLLAddr, IPv6, IPv6ExtHdrFragment, + Padding, Raw, wrpcap) + +if TYPE_CHECKING: + from typing import Callable, Iterator, Optional + + from scapy.packet import Packet + + #: A TCP option list, spelled the way :mod:`scapy` takes it. + Options = list[tuple[str, object]] + +__all__ = ['generate'] + +#: Repository root, i.e. the grandparent of the directory holding this file. +ROOT = pathlib.Path(__file__).resolve().parents[2] +#: Default destination directory for the generated captures. +SAMPLE = ROOT / 'examples' / 'sample' + +#: Capture start time, fixed so that regenerating gives identical files. +EPOCH = 1500000000.0 + + +############################################################################### +# Deterministic filler +############################################################################### + + +def _filler(length: 'int', tag: 'bytes') -> 'bytes': + """Deterministic opaque payload bytes. + + Args: + length: Number of octets required. + tag: Seed distinguishing one payload from another. + + Returns: + Exactly ``length`` octets, the same on every machine and every run. + + """ + out = bytearray() + counter = 0 + while len(out) < length: + out += hashlib.sha256(b'%s/%d' % (tag, counter)).digest() + counter += 1 + return bytes(out[:length]) + + +def _text_filler(length: 'int', tag: 'bytes') -> 'str': + """Deterministic printable filler, for use inside generated source text. + + Args: + length: Number of characters required. + tag: Seed distinguishing one run of filler from another. + + Returns: + Exactly ``length`` characters drawn from the base16 alphabet. + + """ + return _filler((length + 1) // 2, tag).hex()[:length] + + +############################################################################### +# Checksum solving +############################################################################### + + +def _ones_add(left: 'int', right: 'int') -> 'int': + """Add two 16-bit words with end-around carry (one's complement addition). + + Args: + left: First addend. + right: Second addend. + + Returns: + The one's complement sum, as a 16-bit word. + + """ + total = left + right + return (total & 0xffff) + (total >> 16) + + +def _solve_udp_checksum(build: 'Callable[[int], Packet]', target: 'int') -> 'Packet': + """Solve for the free 16-bit word that gives a datagram a wanted checksum. + + The UDP checksum is the one's complement of the one's complement sum of the + datagram and its pseudo header, so a single 16-bit word at an even offset + enters that sum exactly once. Building the packet once with the word zeroed + is therefore enough to compute the word that lands on ``target``. + + Args: + build: Builds the packet, given the value of the free word. + target: Wanted value of the UDP checksum field. + + Returns: + The packet whose (scapy-computed) UDP checksum is ``target``. + + Raises: + RuntimeError: If the solved packet does not carry ``target`` after all, + which would mean ``build`` does not use the word as assumed. + + """ + probe = build(0) + probe = probe.__class__(bytes(probe)) # rebuild so scapy fills the checksum in + partial = 0xffff ^ probe[UDP].chksum # sum of the datagram, word zeroed + word = _ones_add(0xffff ^ target, 0xffff ^ partial) + + packet = build(word) + check = packet.__class__(bytes(packet))[UDP].chksum + if check != target: + raise RuntimeError(f'cannot solve UDP checksum: wanted {target:#06x}, got {check:#06x}') + return packet + + +############################################################################### +# Capture assembly +############################################################################### + + +class _Endpoint(NamedTuple): + """One end of a conversation.""" + + #: Ethernet address. + mac: 'str' + #: IPv4 or IPv6 address. + ip: 'str' + + +class _Capture: + """An ordered list of frames, timestamped as they are appended.""" + + def __init__(self, start: 'float' = EPOCH, step: 'float' = 0.000431) -> None: + """Initialisation. + + Args: + start: Capture time of the first frame. + step: Nominal delay between frames; jittered deterministically so + that the timestamps do not read as machine generated. + + """ + self._frames = [] # type: list[Packet] + self._clock = start + self._step = step + + def __len__(self) -> 'int': + """Number of frames appended so far.""" + return len(self._frames) + + @property + def frames(self) -> 'list[Packet]': + """The frames, in capture order.""" + return self._frames + + def add(self, packet: 'Packet') -> 'Packet': + """Timestamp a packet and append it to the capture. + + Args: + packet: Frame to append. + + Returns: + The packet, as appended. + + """ + index = len(self._frames) + self._clock += self._step + (index % 17) * 0.000037 + packet.time = self._clock + self._frames.append(packet) + return packet + + def write(self, path: 'pathlib.Path') -> 'pathlib.Path': + """Write the capture out, replacing any existing file. + + Args: + path: Destination file. + + Returns: + The path written. + + """ + wrpcap(str(path), self._frames) + return path + + +def _network(src: 'str', dst: 'str', *, hops: 'int', ident: 'int' = 0) -> 'Packet': + """Build the network layer for an address pair, IPv4 or IPv6 as required. + + Args: + src: Source address. + dst: Destination address. + hops: IPv4 time to live, or IPv6 hop limit. + ident: IPv4 identification field; ignored for IPv6. + + Returns: + The network layer, without payload. + + """ + if ':' in src: + return IPv6(src=src, dst=dst, hlim=hops) + return IP(src=src, dst=dst, ttl=hops, id=ident) + + +def _timestamps(value: 'int', echo: 'int') -> 'list[tuple[str, object]]': + """TCP options for a segment carrying nothing but timestamps. + + Args: + value: Timestamp value. + echo: Timestamp echo reply. + + Returns: + Option list, 12 octets once encoded, giving a 32-octet header. + + """ + return [('NOP', None), ('NOP', None), ('Timestamp', (value, echo))] + + +class _Flow: + """A single TCP connection, tracking sequence state for both directions. + + Every method returns one frame and advances the connection state, so a + caller only has to say what happens next; the sequence and acknowledgement + numbers, IPv4 identifications and TCP timestamps follow from that. + + """ + + def __init__(self, client: '_Endpoint', server: '_Endpoint', sport: 'int', + dport: 'int', *, client_window: 'int' = 65535, + server_window: 'int' = 14480, hops: 'int' = 64) -> None: + """Initialisation. + + Args: + client: The side that sends the SYN. + server: The side that listens. + sport: Client (ephemeral) port. + dport: Server port. + client_window: Window advertised by the client. + server_window: Window advertised by the server. + hops: Time to live / hop limit on every frame of the connection. + + """ + self.client = client + self.server = server + self.sport = sport + self.dport = dport + self.client_window = client_window + self.server_window = server_window + self.hops = hops + + # initial sequence numbers and timestamp clocks, spread out per + # connection the way separate connections really are + self._seq = {True: 0x1f2c3d00 + sport * 5779, False: 0x8f3a1200 + sport * 8317} + self._ts = {True: 834459645 + sport * 31, False: 2454851095 + sport * 17} + self._ident = {True: 0x1000 + sport % 0x2000, False: 0x4000 + sport % 0x2000} + + def _segment(self, to_server: 'bool', flags: 'str', *, payload: 'bytes' = b'', + options: 'Optional[Options]' = None, + window: 'Optional[int]' = None) -> 'Packet': + """Build one segment of the connection. + + Args: + to_server: Direction of travel. + flags: TCP flags, in :mod:`scapy` spelling. + payload: Segment data. + options: TCP options; timestamps only, if not given. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame, ready to be appended to a capture. + + """ + source, target = (self.client, self.server) if to_server else (self.server, self.client) + sport, dport = (self.sport, self.dport) if to_server else (self.dport, self.sport) + if window is None: + window = self.client_window if to_server else self.server_window + if options is None: + self._ts[to_server] += 4 + options = _timestamps(self._ts[to_server], self._ts[not to_server]) + + self._ident[to_server] = (self._ident[to_server] + 1) & 0xffff + segment = TCP(sport=sport, dport=dport, flags=flags, seq=self._seq[to_server], + ack=self._seq[not to_server] if 'A' in flags else 0, + window=window, options=options) + + # SYN and FIN each occupy one sequence number, as does every data octet + self._seq[to_server] += len(payload) + (1 if set('SF') & set(flags) else 0) + + frame = (Ether(src=source.mac, dst=target.mac) + / _network(source.ip, target.ip, hops=self.hops, + ident=self._ident[to_server]) + / segment) + return frame / Raw(load=payload) if payload else frame + + def syn(self, *, options: 'Optional[Options]' = None) -> 'Packet': + """Client SYN opening the connection. + + Args: + options: TCP options; the set a Linux client sends, if not given. + + Returns: + The frame. + + """ + if options is None: + options = [('MSS', 1460), ('SAckOK', b''), + ('Timestamp', (self._ts[True], 0)), ('NOP', None), ('WScale', 6)] + return self._segment(True, 'S', options=options) + + def synack(self, *, options: 'Optional[Options]' = None) -> 'Packet': + """Server SYN-ACK answering the client's SYN. + + Args: + options: TCP options; the set a Linux server sends, if not given. + + Returns: + The frame. + + """ + if options is None: + options = [('MSS', 1460), ('SAckOK', b''), + ('Timestamp', (self._ts[False], self._ts[True])), + ('NOP', None), ('WScale', 7)] + return self._segment(False, 'SA', options=options) + + def ack(self, to_server: 'bool' = True, *, options: 'Optional[Options]' = None, + window: 'Optional[int]' = None) -> 'Packet': + """A pure acknowledgement, carrying no data. + + Args: + to_server: Direction of travel. + options: TCP options; timestamps only, if not given. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame. + + """ + return self._segment(to_server, 'A', options=options, window=window) + + def data(self, to_server: 'bool', payload: 'bytes', *, + options: 'Optional[Options]' = None, window: 'Optional[int]' = None) -> 'Packet': + """A segment carrying data, with the push flag set. + + Args: + to_server: Direction of travel. + payload: Segment data. + options: TCP options; timestamps only, if not given. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame. + + """ + return self._segment(to_server, 'PA', payload=payload, options=options, window=window) + + def fin(self, to_server: 'bool' = True, *, options: 'Optional[Options]' = None, + window: 'Optional[int]' = None) -> 'Packet': + """A FIN-ACK, closing one direction of the connection. + + Args: + to_server: Direction of travel. + options: TCP options; timestamps only, if not given. + window: Advertised window; the endpoint's default, if not given. + + Returns: + The frame. + + """ + return self._segment(to_server, 'FA', options=options, window=window) + + +############################################################################### +# examples/sample/arp.pcap +############################################################################### + + +def _write_arp(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``arp.pcap``. + + A host refreshes a neighbour it already has cached, so the request is + unicast and already carries the target hardware address, and the neighbour + answers. Both frames are padded to the 60-octet Ethernet minimum. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + local = _Endpoint('00:0c:29:19:dc:61', '10.20.30.131') + peer = _Endpoint('00:0c:29:7d:1d:b4', '10.20.30.130') + capture = _Capture(step=0.000206) + + capture.add(Ether(src=local.mac, dst=peer.mac) + / ARP(op='who-has', hwsrc=local.mac, psrc=local.ip, + hwdst=peer.mac, pdst=peer.ip) + / Padding(load=bytes(18))) + capture.add(Ether(src=peer.mac, dst=local.mac) + / ARP(op='is-at', hwsrc=peer.mac, psrc=peer.ip, + hwdst=local.mac, pdst=local.ip) + / Padding(load=bytes(18))) + + return capture.write(dest / 'arp.pcap') + + +############################################################################### +# examples/sample/ipv4.pcap +############################################################################### + + +def _write_ipv4(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``ipv4.pcap``. + + Four datagrams of a local-scope IPv4 multicast stream. The payload is + opaque stream data; its final hextet is the free word solved for in the + first datagram, whose checksum the tests pin. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + source = '172.31.127.230' + group = '239.1.3.3' + capture = _Capture(step=0.001284) + + def build(index: 'int', word: 'int') -> 'Packet': + payload = _filler(1826, b'ipv4-multicast/%d' % index) + word.to_bytes(2, 'big') + return (Ether(src='00:0c:29:3f:1a:07', dst='01:00:5e:01:03:03') + / IP(src=source, dst=group, ttl=1, id=0x7a00 + index) + / UDP(sport=42054, dport=12345) + / Raw(load=payload)) + + capture.add(_solve_udp_checksum(lambda word: build(0, word), 0xff0b)) + for index in range(1, 4): + capture.add(build(index, 0x0000)) + + return capture.write(dest / 'ipv4.pcap') + + +############################################################################### +# examples/sample/ipv6.pcap +############################################################################### + + +def _write_ipv6(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``ipv6.pcap``. + + Neighbour discovery and ICMPv6 echoes between two link-local hosts, + followed by a 4778-octet UDP datagram fragmented to fit a 1500-octet MTU. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + local = _Endpoint('00:0c:29:aa:b1:5e', 'fe80::a423:b61d:7c92:70c6') + peer = _Endpoint('80:1f:12:c9:d1:3d', 'fe80::821f:12ff:fec9:d13d') + capture = _Capture(step=0.000912) + + def solicit(dst_mac: 'str', dst_ip: 'str') -> 'Packet': + return (Ether(src=local.mac, dst=dst_mac) + / IPv6(src=local.ip, dst=dst_ip, hlim=255) + / ICMPv6ND_NS(tgt=peer.ip) + / ICMPv6NDOptSrcLLAddr(lladdr=local.mac)) + + def advertise() -> 'Packet': + return (Ether(src=peer.mac, dst=local.mac) + / IPv6(src=peer.ip, dst=local.ip, hlim=255) + / ICMPv6ND_NA(tgt=peer.ip, R=0, S=1, O=1) + / ICMPv6NDOptDstLLAddr(lladdr=peer.mac)) + + # 0-1: the peer is resolved for the first time, via the solicited-node group + capture.add(solicit('33:33:ff:c9:d1:3d', 'ff02::1:ffc9:d13d')) + capture.add(advertise()) + + # 2-3: and probed again once the cache entry goes stale, this time unicast + capture.add(solicit(peer.mac, peer.ip)) + capture.add(advertise()) + + # 4-11: four echo exchanges, to show a plain ICMPv6 payload + for index in range(4): + echo = _filler(56, b'ipv6-echo/%d' % index) + capture.add(Ether(src=local.mac, dst=peer.mac) + / IPv6(src=local.ip, dst=peer.ip, hlim=64) + / ICMPv6EchoRequest(id=0x3f21, seq=index, data=echo)) + capture.add(Ether(src=peer.mac, dst=local.mac) + / IPv6(src=peer.ip, dst=local.ip, hlim=64) + / ICMPv6EchoReply(id=0x3f21, seq=index, data=echo)) + + # 12-15: one 4778 octet datagram, fragmented for a 1500 octet MTU + datagram = bytes(IPv6(src=local.ip, dst=peer.ip) + / UDP(sport=51234, dport=5001) + / Raw(load=_filler(4770, b'ipv6-bulk')))[40:] + offset = 0 + while offset < len(datagram): + chunk = datagram[offset:offset + 1448] + more = offset + len(chunk) < len(datagram) + capture.add(Ether(src=local.mac, dst=peer.mac) + / IPv6(src=local.ip, dst=peer.ip, hlim=64, nh=44) + / IPv6ExtHdrFragment(nh=17, offset=offset // 8, + m=1 if more else 0, id=110308) + / Raw(load=chunk)) + offset += len(chunk) + + return capture.write(dest / 'ipv6.pcap') + + +############################################################################### +# examples/sample/tcp.pcap +############################################################################### + + +def _write_tcp(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``tcp.pcap``. + + An excerpt of two SSH sessions to the same neighbour, one over IPv4 and + one over IPv6, that begins with the IPv4 server's SYN-ACK: the client's + SYN is older than the excerpt. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + client4 = _Endpoint('00:0c:29:19:dc:61', '10.20.30.131') + server4 = _Endpoint('00:0c:29:7d:1d:b4', '10.20.30.130') + client6 = _Endpoint('00:0c:29:19:dc:61', 'fe80::a6:87f9:2793:16ee') + server6 = _Endpoint('00:0c:29:7d:1d:b4', 'fe80::1ccd:7c77:bac7:46b7') + capture = _Capture(step=0.000774) + + over_v4 = _Flow(client4, server4, 53406, 22, server_window=65535) + over_v6 = _Flow(client6, server6, 51774, 22, client_window=4096, server_window=8192) + + # the client's SYN predates the excerpt, so its sequence number is only + # known to the flow once the server has acknowledged it + over_v4.syn() + + # 0: the server's SYN-ACK, offering the full option set a BSD stack does: + # maximum segment size, window scale, timestamps, SACK permitted, and + # an explicit end of option list padded out to a 44 octet header + capture.add(over_v4.synack(options=[ + ('MSS', 1460), ('NOP', None), ('WScale', 6), ('NOP', None), ('NOP', None), + ('Timestamp', (834459645, 2454851095)), ('SAckOK', b''), ('EOL', None), + ])) + # 1: the client completes the handshake + capture.add(over_v4.ack()) + + # 2-4: meanwhile the IPv6 session exchanges its version banners + over_v6.syn() + over_v6.synack() + over_v6.ack() + capture.add(over_v6.data(True, b'SSH-2.0-OpenSSH_9.6\r\n')) + capture.add(over_v6.ack(False, options=_timestamps(1906342664, 2559889017))) + capture.add(over_v6.data(False, b'SSH-2.0-OpenSSH_9.3\r\n')) + + # 5-6: an encrypted record on the IPv4 session, and its acknowledgement + capture.add(over_v4.data(True, _filler(212, b'tcp-ssh-record'))) + capture.add(over_v4.ack(False)) + + return capture.write(dest / 'tcp.pcap') + + +############################################################################### +# examples/sample/stream.pcap +############################################################################### + +#: Service types a Bonjour client looks for when hunting media receivers. The +#: encoded question section is exactly 125 octets, giving a 137 octet query. +_MDNS_SERVICES = ('_googlecast._tcp.local', '_ipp._tcp.local', '_ipps._tcp.local', + '_raop._tcp.local', '_companion-link._tcp.local') + + +def _write_stream(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``stream.pcap``. + + An excerpt of a media stream over link-local IPv6, with a multicast DNS + service query from a third host on the link interleaved into it. The + querier's address carries the free word solved for to give the query the + checksum the tests pin; nothing else in the capture refers to it. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + client = _Endpoint('00:0c:29:19:dc:61', 'fe80::a6:87f9:2793:16ee') + server = _Endpoint('80:1f:12:c9:d1:3d', 'fe80::821f:12ff:fec9:d13d') + capture = _Capture(step=0.002317) + + flow = _Flow(client, server, 49312, 7000, client_window=45, server_window=2048) + flow.syn() + flow.synack() + flow.ack() + + def query(word: 'int') -> 'Packet': + return (Ether(src='9c:20:7b:6e:14:aa', dst='33:33:00:00:00:fb') + / IPv6(src='fe80::b8f3:%04x:5d0e:71c2' % word, dst='ff02::fb', hlim=255) + / UDP(sport=5353, dport=5353) + / DNS(id=0, qr=0, rd=0, + qd=[DNSQR(qname=name, qtype='PTR', qclass=1) + for name in _MDNS_SERVICES])) + + # 0: the stream is already running when the excerpt starts + capture.add(flow.data(True, _filler(96, b'stream-setup'))) + # 1: a neighbour on the link goes looking for media receivers + capture.add(_solve_udp_checksum(query, 0x9cb1)) + # 2-4: a request from the client, acknowledged, then answered + capture.add(flow.data(True, _filler(72, b'stream-request'))) + capture.add(flow.ack(False, window=2043, options=_timestamps(236969253, 1144394525))) + capture.add(flow.data(False, _filler(1024, b'stream-media'))) + # 5: and acknowledged in turn + capture.add(flow.ack()) + + return capture.write(dest / 'stream.pcap') + + +############################################################################### +# examples/sample/http.pcap +############################################################################### + +#: Hosts the generated page load talks to, and the address each resolved to. +_HTTP_HOSTS = { + 'd1.sina.com.cn': '118.144.75.171', + 'd7.sina.com.cn': '118.144.75.177', + 'www.sina.com.cn': '180.149.138.243', + 'sports.sina.com.cn': '180.149.138.100', + 'beacon.sina.com.cn': '123.126.104.19', +} + +#: Resources the background connections fetch, as ``(host, path, type)``. +_HTTP_RESOURCES = ( + ('www.sina.com.cn', '/css/global.css', 'text/css'), + ('www.sina.com.cn', '/js/index.js', 'application/x-javascript'), + ('d1.sina.com.cn', '/litong/zhitou/sinaads/release/sinaads_ck.js', 'application/x-javascript'), + ('d7.sina.com.cn', '/litong/zhitou/sinaads/demo/wenjing8/style.css', 'text/css'), + ('sports.sina.com.cn', '/css/sports_index.css', 'text/css'), + ('www.sina.com.cn', '/images/logo_sina.gif', 'image/gif'), + ('sports.sina.com.cn', '/js/nba_scoreboard.js', 'application/x-javascript'), + ('d1.sina.com.cn', '/litong/zhitou/sinaads/demo/wenjing8/banner.gif', 'image/gif'), +) + +#: Client and gateway of the capture, which the ARP sample resolves for. +_HTTP_CLIENT = _Endpoint('00:0c:29:19:dc:61', '10.20.30.131') +_HTTP_GATEWAY = '00:50:56:c0:00:08' + + +def _http_request(host: 'str', path: 'str', *, referer: 'Optional[str]' = None, + close: 'bool' = False) -> 'bytes': + """Build an HTTP/1.1 GET request as a browser would send it. + + Args: + host: Value of the ``Host`` header. + path: Request target. + referer: Value of the ``Referer`` header, if any. + close: Whether to ask for the connection to be closed. + + Returns: + The encoded request. + + """ + lines = [ + 'GET %s HTTP/1.1' % path, + 'Host: %s' % host, + 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/601.7.7 ' + '(KHTML, like Gecko) Version/9.1.2 Safari/601.7.7', + 'Accept: */*', + 'Accept-Language: zh-cn', + 'Accept-Encoding: gzip, deflate', + ] + if referer is not None: + lines.append('Referer: %s' % referer) + lines.append('Connection: close' if close else 'Connection: keep-alive') + return ('\r\n'.join(lines) + '\r\n\r\n').encode() + + +def _http_response(body: 'bytes', *, server: 'str' = 'nginx', content_type: 'str' = 'text/html', + status: 'str' = '200 OK', length: 'Optional[int]' = None, + encoding: 'Optional[str]' = None, close: 'bool' = False) -> 'bytes': + """Build an HTTP/1.1 response, header block and body together. + + Args: + body: Body octets to append after the header block. + server: Value of the ``Server`` header. + content_type: Value of the ``Content-Type`` header. + status: Status line, code and reason phrase. + length: Value of the ``Content-Length`` header; the body's own length, + if not given. Pass it explicitly when the body continues into + later segments. + encoding: Value of the ``Content-Encoding`` header, if any. + close: Whether the server is announcing it will close. + + Returns: + The encoded response. + + """ + lines = [ + 'HTTP/1.1 %s' % status, + 'Server: %s' % server, + 'Date: Fri, 14 Jul 2017 02:40:00 GMT', + 'Content-Type: %s' % content_type, + 'Content-Length: %d' % (len(body) if length is None else length), + ] + if encoding is not None: + lines.append('Content-Encoding: %s' % encoding) + lines.append('Connection: close' if close else 'Connection: keep-alive') + return ('\r\n'.join(lines) + '\r\n\r\n').encode() + body + + +def _gzip_body(size: 'int') -> 'bytes': + """Compress generated JavaScript to a gzip stream of an exact size. + + The response the tests read announces ``Content-Length: 3520``, so the + body has to be that long. The source text is padded with incompressible + filler until the compressed stream lands on it exactly, which keeps the + body a real gzip stream that decompresses to real text. + + Args: + size: Wanted length of the gzip stream. + + Returns: + A gzip stream of exactly ``size`` octets. + + Raises: + RuntimeError: If no amount of padding gives that length, which would + mean the local :mod:`zlib` compresses in coarser steps than this + search assumes. + + """ + template = ( + '/*! sinaads toutiaobaoMedia %s */\n' + '(function(w,d){var C={id:"toutiaobao",slot:"media",ver:"1.4.1"},' + 'T=["zaowanbao","wenjing8","media"],K="%s";\n' + 'function esc(s){return encodeURIComponent(String(s))}\n' + 'function url(p){return"//d1.sina.com.cn/litong/zhitou/sinaads/"+p+"?k="+esc(K)}\n' + 'function load(p,cb){var s=d.createElement("script");s.async=1;s.src=url(p);' + 's.onload=cb;d.body.appendChild(s)}\n' + 'w.sinaads=w.sinaads||[];w.sinaads.push({conf:C,tags:T,load:load,pad:"%s"});\n' + '})(window,document);\n' + ) + for extra in range(8): + for pad in range(0, 6144): + source = template % (_text_filler(8, b'js-build'), + _text_filler(24, b'js-key'), + _text_filler(pad, b'js-pad') + ' ' * extra) + blob = gzip.compress(source.encode(), compresslevel=9, mtime=0) + if len(blob) == size: + return blob + if len(blob) > size: + break + raise RuntimeError(f'cannot build a gzip body of exactly {size} octets') + + +class _HTTPBuilder: + """Assembles ``http.pcap``, pacing it so pinned frames land on their index. + + The HTTP tests index frames directly, so this class keeps a stream of + background connections on tap: :meth:`fill` runs it until the capture is + one frame short of a wanted index, and the caller then appends the frame + the test is looking for. + + """ + + def __init__(self) -> None: + """Initialisation.""" + self.capture = _Capture(step=0.000638) + self._port = 53500 + self._background = self._traffic() + + def port(self) -> 'int': + """Allocate the next ephemeral port of the capture.""" + self._port += 1 + return self._port + + def flow(self, host: 'str', *, sport: 'Optional[int]' = None) -> '_Flow': + """Open a connection to one of the hosts of the capture. + + Args: + host: Host name, which must be one of :data:`_HTTP_HOSTS`. + sport: Client port; the next one allocated, if not given. + + Returns: + The connection, before its handshake. + + """ + server = _Endpoint(_HTTP_GATEWAY, _HTTP_HOSTS[host]) + return _Flow(_HTTP_CLIENT, server, self.port() if sport is None else sport, 80, + server_window=14480, hops=64) + + def add(self, packet: 'Packet') -> 'Packet': + """Append a frame to the capture.""" + return self.capture.add(packet) + + def fill(self, index: 'int') -> 'None': + """Run background traffic until the next frame appended lands on ``index``. + + Args: + index: Frame index the caller is about to fill. + + Raises: + RuntimeError: If the capture has already passed ``index``, which + means the layout below no longer adds up. + + """ + if len(self.capture) > index: + raise RuntimeError(f'frame {index} is already taken by {len(self.capture)} frames') + while len(self.capture) < index: + self.capture.add(next(self._background)) + + def _traffic(self) -> 'Iterator[Packet]': + """Yield the frames of an unending series of short HTTP connections. + + Each connection is complete in itself -- handshake, one request, one + response small enough to fit a single segment, and an orderly close -- + so the traffic surrounding the pinned frames is as well formed as the + pinned frames themselves. + + """ + index = 0 + while True: + host, path, content_type = _HTTP_RESOURCES[index % len(_HTTP_RESOURCES)] + flow = self.flow(host) + body = _filler(96 + (index % 7) * 32, b'http-filler/%d' % index) + + yield flow.syn() + yield flow.synack() + yield flow.ack() + yield flow.data(True, _http_request(host, '%s?v=%d' % (path, index), close=True)) + yield flow.ack(False) + yield flow.data(False, _http_response(body, content_type=content_type, close=True)) + yield flow.ack() + yield flow.fin() + yield flow.fin(False) + yield flow.ack() + + index += 1 + + +def _write_http(dest: 'pathlib.Path') -> 'pathlib.Path': + """Write ``http.pcap``. + + The exchanges below are the ones the tests read, each paced onto the frame + index it is read at; everything between them is background traffic from + :meth:`_HTTPBuilder._traffic`. + + Args: + dest: Destination directory. + + Returns: + The path written. + + """ + builder = _HTTPBuilder() + home = 'http://www.sina.com.cn/' + + # ------------------------------------------------------------------ + # frames 114 and 117: a gzip'd script from d1.sina.com.cn, whose body + # spans three segments + # ------------------------------------------------------------------ + script = builder.flow('d1.sina.com.cn', sport=53406) + body = _gzip_body(3520) + builder.fill(110) + builder.add(script.syn()) + builder.add(script.synack()) + builder.add(script.ack()) + + builder.fill(114) + builder.add(script.data(True, _http_request( + 'd1.sina.com.cn', + '/litong/zhitou/sinaads/demo/wenjing8/ZaoWanBao/toutiaobaoMedia.js', + referer=home))) + builder.add(script.ack(False)) + + builder.fill(117) + builder.add(script.data(False, _http_response( + body[:1360], server='Tengine', content_type='application/x-javascript', + length=len(body), encoding='gzip'))) + builder.add(script.ack()) + builder.add(script.data(False, body[1360:2720])) + builder.add(script.ack()) + builder.add(script.data(False, body[2720:])) + builder.add(script.ack()) + builder.add(script.fin(False)) + builder.add(script.fin()) + builder.add(script.ack(False)) + + # ------------------------------------------------------------------ + # frame 393: the sports front page + # ------------------------------------------------------------------ + sports = builder.flow('sports.sina.com.cn', sport=53410) + builder.fill(388) + builder.add(sports.syn()) + builder.add(sports.synack()) + builder.add(sports.ack()) + + builder.fill(393) + builder.add(sports.data(True, _http_request('sports.sina.com.cn', '/', referer=home))) + builder.add(sports.ack(False)) + + builder.fill(397) + builder.add(sports.data(False, _http_response( + _filler(512, b'http-sports-index'), server='nginx/1.10.3', + content_type='text/html; charset=utf-8'))) + builder.add(sports.ack()) + builder.add(sports.fin()) + builder.add(sports.fin(False)) + builder.add(sports.ack()) + + # ------------------------------------------------------------------ + # frame 556: the shared advertisement script from d7.sina.com.cn + # ------------------------------------------------------------------ + sinaads = builder.flow('d7.sina.com.cn', sport=53412) + builder.fill(550) + builder.add(sinaads.syn()) + builder.add(sinaads.synack()) + builder.add(sinaads.ack()) + + builder.fill(556) + builder.add(sinaads.data(True, _http_request( + 'd7.sina.com.cn', '/litong/zhitou/sinaads/release/sinaads.js', referer=home))) + builder.add(sinaads.ack(False)) + + builder.fill(560) + builder.add(sinaads.data(False, _http_response( + _filler(880, b'http-sinaads-js'), server='Tengine', + content_type='application/x-javascript'))) + builder.add(sinaads.ack()) + builder.add(sinaads.fin()) + builder.add(sinaads.fin(False)) + builder.add(sinaads.ack()) + + # ------------------------------------------------------------------ + # frame 587: a beacon acknowledged with a 35 octet JSON body + # ------------------------------------------------------------------ + beacon = builder.flow('beacon.sina.com.cn', sport=53414) + acknowledged = b'{"ret":0,"msg":"ok","id":"a1b2c3"}\n' + if len(acknowledged) != 35: + raise RuntimeError(f'beacon body is {len(acknowledged)} octets, expected 35') + builder.fill(578) + builder.add(beacon.syn()) + builder.add(beacon.synack()) + builder.add(beacon.ack()) + + builder.fill(583) + builder.add(beacon.data(True, _http_request( + 'beacon.sina.com.cn', '/a.gif?noScriptFlag=1&referrer=%s' % home, referer=home))) + builder.add(beacon.ack(False)) + + builder.fill(587) + builder.add(beacon.data(False, _http_response( + acknowledged, server='Suda/1.12.0', content_type='application/json'))) + builder.add(beacon.ack()) + builder.add(beacon.fin()) + builder.add(beacon.fin(False)) + builder.add(beacon.ack()) + + # ------------------------------------------------------------------ + # frame 629: a beacon acknowledged with no body at all + # ------------------------------------------------------------------ + empty = builder.flow('beacon.sina.com.cn', sport=53416) + builder.fill(620) + builder.add(empty.syn()) + builder.add(empty.synack()) + builder.add(empty.ack()) + + builder.fill(625) + builder.add(empty.data(True, _http_request( + 'beacon.sina.com.cn', '/a.gif?dpc=1&referrer=%s' % home, referer=home))) + builder.add(empty.ack(False)) + + builder.fill(629) + builder.add(empty.data(False, _http_response( + b'', server='Suda/1.12.0', content_type='image/gif'))) + builder.add(empty.ack()) + builder.add(empty.fin()) + builder.add(empty.fin(False)) + builder.add(empty.ack()) + + # ------------------------------------------------------------------ + # frame 1113: a continuation segment, carrying body octets and no header + # block, of an image large enough to need four segments + # ------------------------------------------------------------------ + image = builder.flow('www.sina.com.cn', sport=53418) + picture = _filler(4200, b'http-picture') + builder.fill(1100) + builder.add(image.syn()) + builder.add(image.synack()) + builder.add(image.ack()) + + builder.fill(1105) + builder.add(image.data(True, _http_request( + 'www.sina.com.cn', '/images/2017/0714/headline_640x360.jpg', referer=home))) + builder.add(image.ack(False)) + + builder.fill(1108) + builder.add(image.data(False, _http_response( + picture[:1200], server='nginx', content_type='image/jpeg', length=len(picture)))) + builder.add(image.ack()) + builder.add(image.data(False, picture[1200:2400])) + builder.add(image.ack()) + builder.add(image.data(False, picture[2400:3600])) + builder.add(image.data(False, picture[3600:])) + builder.add(image.ack()) + builder.add(image.fin(False)) + builder.add(image.fin()) + + return builder.capture.write(dest / 'http.pcap') + + +############################################################################### +# Entry point +############################################################################### + + +def generate(dest: 'pathlib.Path | None' = None) -> 'list[pathlib.Path]': + """Write the ``.pcap`` sample fixtures. + + Args: + dest: Destination directory; ``examples/sample/`` under the repository + root, if not given. Created if it does not exist. + + Returns: + The paths written, in the order they were written. + + """ + dest = SAMPLE if dest is None else pathlib.Path(dest) + dest.mkdir(parents=True, exist_ok=True) + + return [ + _write_arp(dest), + _write_ipv4(dest), + _write_ipv6(dest), + _write_tcp(dest), + _write_stream(dest), + _write_http(dest), + ] + + +if __name__ == '__main__': + from scapy.utils import rdpcap + + for sample in generate(): + print('%-24s %8d octets %5d frames' % ( + sample.relative_to(ROOT), sample.stat().st_size, len(rdpcap(str(sample))))) diff --git a/examples/samples/pcapng.py b/examples/samples/pcapng.py new file mode 100644 index 0000000000..23d439faa4 --- /dev/null +++ b/examples/samples/pcapng.py @@ -0,0 +1,949 @@ +# -*- coding: utf-8 -*- +"""Provision the ``.pcapng`` sample fixtures used by the test suite. + +The test suite (``tests/protocols/test_pcapng_regression.py``) reads five +PCAP-NG captures out of ``examples/sample/`` that :file:`.gitignore` deliberately keeps +out of the repository. This module puts them back on any machine, without a +checkout of anything private, by two routes: + +* **Downloaded** -- fetched from a public source, with the source URL and the + SHA-256 of the exact bytes pinned below. The digest is checked after every + download and a mismatch is a hard error, never a warning. If the network is + unavailable the download degrades to a synthesised stand-in, and both the + console output and the return value say so rather than pretending otherwise. +* **Synthesised** -- written byte by byte by :class:`_Blocks` below, with no + network access whatsoever, from packet payloads carried in the committed + ``examples/sample/dhcp.pcapng`` fixture. Generation is deterministic: the same input + tree always produces the same bytes, so a second run is a no-op. + +Which fixture comes from where: + +=========================== ============= =================================== +Fixture Provenance Exercises +=========================== ============= =================================== +``dhcp_big_endian.pcapng`` downloaded big-endian section header block +``dhcp_little_endian.pcapng`` synthesised little-endian section header block +``many_interfaces.pcapng`` downloaded eleven interface description blocks +``test.pcapng`` synthesised the auxiliary block types +``profile.pcapng`` synthesised the ``if_*``/``isb_*`` option space +=========================== ============= =================================== + +Provenance and licensing of the downloaded captures +--------------------------------------------------- + +Both downloads come from the Wireshark source tree's ``test/captures`` +directory (https://gitlab.com/wireshark/wireshark, mirrored on GitHub), which +is distributed under the GNU GPL v2 or later. They are fetched into ``examples/sample/``, +which :file:`.gitignore` excludes, so this project never redistributes them -- +each machine fetches its own copy. ``many_interfaces.pcapng`` does not exist +upstream under that name; upstream ships a three-file ring-buffer set, and the +first member (``many_interfaces.pcapng.1``) is a complete, self-contained +PCAP-NG file, so that is what is fetched and stored under the name the test +expects. + +The committed ``examples/sample/dhcp.pcapng`` is byte-identical to upstream's +``test/captures/dhcp.pcapng`` (SHA-256 ``e47f667c...5a1666``), which is where +the synthesised DHCP payloads come from. + +Block types deliberately left out +--------------------------------- + +Three constructs that :mod:`pcapkit.protocols.misc.pcapng` nominally supports +raise on spec-conformant input, so no fixture here contains them -- a fixture +that cannot be parsed at all is of no use to a regression test: + +* **Custom Block** (``0x00000BAD``/``0x40000BAD``) -- + ``pcapkit/protocols/schema/misc/pcapng.py:1591`` computes its padding as + ``(4 - pkt['data'] % 4) % 4`` where ``pkt['data']`` is :class:`bytes`, so any + custom block raises ``TypeError: not all arguments converted during bytes + formatting``. The ``len()`` call is missing. +* **Packet Block** (obsolete, ``0x00000002``) -- its ``interface_id`` and + ``drop_count`` are declared ``UInt32Field`` at lines 1657 and 1659, but the + format spec makes both 16-bit, and the ``options`` length at line 1674 + subtracts the 32-byte prefix that only the 16-bit layout produces. The field + widths and the arithmetic disagree, so neither layout parses. +* **``if_IPv6addr``** -- ``IPv6InterfaceField.post_process`` in + ``pcapkit/corekit/fields/ipaddress.py`` reads the trailing prefix-length + octet as ``int(value[16:])``, i.e. it parses a raw byte as an ASCII decimal + string, so a ``/64`` prefix raises ``ValueError``. + +A fourth defect is worked around rather than avoided. A **Simple Packet Block** +in a section declaring more than one interface raises ``FormatError: PCAP-NG: +[SPB] invalid section with 2 interfaces`` at +``pcapkit/foundation/engines/pcapng.py:218``, which tests +``len(interfaces) != 1``. The format specification (section 4.4 of +draft-ietf-opsawg-pcapng) permits it -- "in a Section that has more than one +interface, only packets received or transmitted on the interface described by +the first Interface Description Block can be contained in a Simple Packet +Block" -- so the correct check is for *zero* interfaces, not for more than one. +``test.pcapng`` therefore carries its simple packet block in its +single-interface second section, which keeps the block type covered. + +Three further defects are exercised on purpose, because they warn rather than +raise, and a fixture that covers the code path is what will catch a future +crash there. All three are option-area sizing errors in +``pcapkit/protocols/schema/misc/pcapng.py``, and all three were confirmed by +bisecting one block type at a time: + +* **Interface Statistics Block** -- ``options`` is sized ``length - 20`` at line + 1335, but the block's fixed fields occupy 24 octets, so the option area + over-runs into the trailing block length. Unconditional: an explicit + ``opt_endofopt`` does not save it, because the field still consumes its + declared width. Symptom, on every capture containing one including the + downloaded ``many_interfaces.pcapng``: ``packet length < 0: -8`` and + ``[Block 5] block length mismatch: N != 0``. +* **Name Resolution Block options** -- ``options`` is sized + ``__option_padding__ - 4`` at line 1177, which over-runs whenever ``ns_*`` + options are present. Symptom: ``[Block 4] block length mismatch: 60 != 314``. +* **Name Resolution Block IPv6 records** -- ``resol`` is sized ``length - 4`` at + line 1102, copied from the IPv4 record where the address is four octets wide; + an IPv6 address is sixteen, so the name field over-reads by twelve and + swallows the record terminator and the block's options. Symptom: ``packet + length < 0: -29273``, and the block's ``ns_*`` options vanish. + +Every fixture here was checked against an independent PCAP-NG implementation +(scapy's ``PcapNgReader``) as well as against pcapkit, and against a structural +walk asserting that each block's total length is 4-octet aligned, repeated +identically at both ends, and that the block chain covers the file exactly. So +the complaints above are pcapkit's readings, not malformed fixtures. + +One complaint is expected rather than a defect: ``test.pcapng`` carries a +deliberately truncated packet (``captured_len`` below ``original_len``, to +exercise snapshot-length handling) and dissecting it reports ``packet length < +0: -2``, which is simply what a snapped frame looks like to a dissector. + +Usage +----- + +.. code-block:: shell + + python examples/samples/pcapng.py # write into examples/sample + python examples/samples/pcapng.py /tmp/fix # write somewhere else + +""" + +from __future__ import annotations + +import hashlib +import logging +import pathlib +import struct +import sys +import urllib.error +import urllib.request +from typing import TYPE_CHECKING, NamedTuple + +if TYPE_CHECKING: + from typing import Callable, Optional + +__all__ = ['generate'] + +#: Repository root, i.e. the grandparent of the directory holding this script. +ROOT = pathlib.Path(__file__).resolve().parents[2] +#: Default destination directory for the fixtures. +SAMPLE_DIR = ROOT / 'examples' / 'sample' +#: Committed fixture the synthesised DHCP payloads are lifted from. +DHCP_SOURCE = SAMPLE_DIR / 'dhcp.pcapng' + +#: Upstream directory the downloaded captures come from. +WIRESHARK_CAPTURES = ('https://raw.githubusercontent.com/wireshark/wireshark/' + 'master/test/captures/') + +#: Timeout, in seconds, for a single download attempt. +TIMEOUT = 30 + +#: Fixed base timestamp (2017-07-14T02:40:00Z) so generation is reproducible. +BASE_SECONDS = 1_500_000_000 + +# Block type numbers, from the PCAP-NG specification. +BLOCK_SHB = 0x0A0D_0D0A +BLOCK_IDB = 0x0000_0001 +BLOCK_SPB = 0x0000_0003 +BLOCK_NRB = 0x0000_0004 +BLOCK_ISB = 0x0000_0005 +BLOCK_EPB = 0x0000_0006 +BLOCK_JOURNAL = 0x0000_0009 +BLOCK_DSB = 0x0000_000A + +#: Byte order magic, written in the section's own byte order. +BYTEORDER_MAGIC = 0x1A2B_3C4D + +# Option codes. The numbers repeat across namespaces on purpose: the spec scopes +# them to the block they appear in. +OPT_ENDOFOPT = 0 +OPT_COMMENT = 1 +SHB_HARDWARE, SHB_OS, SHB_USERAPPL = 2, 3, 4 +(IF_NAME, IF_DESCRIPTION, IF_IPV4ADDR, IF_IPV6ADDR, IF_MACADDR, IF_EUIADDR, + IF_SPEED, IF_TSRESOL, IF_TZONE, IF_FILTER, IF_OS, IF_FCSLEN, IF_TSOFFSET, + IF_HARDWARE, IF_TXSPEED, IF_RXSPEED) = range(2, 18) +(EPB_FLAGS, EPB_HASH, EPB_DROPCOUNT, EPB_PACKETID, EPB_QUEUE, + EPB_VERDICT) = range(2, 8) +NS_DNSNAME, NS_DNSIP4ADDR, NS_DNSIP6ADDR = 2, 3, 4 +(ISB_STARTTIME, ISB_ENDTIME, ISB_IFRECV, ISB_IFDROP, ISB_FILTERACCEPT, + ISB_OSDROP, ISB_USRDELIV) = range(2, 9) + +#: Name resolution record types. +NRB_RECORD_END, NRB_RECORD_IPV4, NRB_RECORD_IPV6 = 0, 1, 2 + +#: ``TLSK``, the decryption secrets type for a TLS key log. +SECRETS_TLS_KEY_LOG = 0x544C_534B + +#: Link types used by the synthesised fixtures. +LINKTYPE_ETHERNET = 1 +LINKTYPE_RAW = 101 + + +class _Blocks: + """Byte-order aware PCAP-NG block writer. + + Every block is emitted with its body padded to a 4-octet boundary and its + total length repeated at both ends, as the specification requires, so the + fixtures are correct by construction rather than by inspection. + + Args: + endian: :mod:`struct` byte order character, ``'<'`` or ``'>'``. + + """ + + def __init__(self, endian: 'str' = '<') -> 'None': + if endian not in ('<', '>'): + raise ValueError(f'byte order must be < or >, not {endian!r}') + self.endian = endian + + @property + def name(self) -> 'str': + """Human-readable byte order, for console output.""" + return 'little-endian' if self.endian == '<' else 'big-endian' + + def pack(self, fmt: 'str', *args: 'int') -> 'bytes': + """Pack ``args`` in this writer's byte order.""" + return struct.pack(self.endian + fmt, *args) + + @staticmethod + def pad(data: 'bytes') -> 'bytes': + """Pad ``data`` with NULs up to a 4-octet boundary.""" + return data + b'\x00' * (-len(data) % 4) + + def option(self, code: 'int', value: 'bytes') -> 'bytes': + """Encode one option as code, length, value, padding.""" + return self.pack('HH', code, len(value)) + self.pad(value) + + def options(self, items: 'list[tuple[int, bytes]]') -> 'bytes': + """Encode an option list, terminated by ``opt_endofopt``. + + An empty list encodes to nothing at all: the specification makes the + whole option area optional, and a bare ``opt_endofopt`` is not the same + thing as its absence. + + """ + if not items: + return b'' + return b''.join(self.option(code, value) for code, value in items) \ + + self.pack('HH', OPT_ENDOFOPT, 0) + + def block(self, block_type: 'int', body: 'bytes') -> 'bytes': + """Wrap ``body`` in a block header and trailer.""" + body = self.pad(body) + total = len(body) + 12 + return self.pack('II', block_type, total) + body + self.pack('I', total) + + # -- individual block types ------------------------------------------------ + + def shb(self, options: 'Optional[list[tuple[int, bytes]]]' = None) -> 'bytes': + """Section Header Block. + + The byte order magic is written in this writer's byte order, which is + exactly how a reader is meant to discover the section's byte order. + + """ + body = self.pack('IHHq', BYTEORDER_MAGIC, 1, 0, -1) + return self.block(BLOCK_SHB, body + self.options(options or [])) + + def idb(self, linktype: 'int' = LINKTYPE_ETHERNET, snaplen: 'int' = 0x0004_0000, + options: 'Optional[list[tuple[int, bytes]]]' = None) -> 'bytes': + """Interface Description Block.""" + body = self.pack('HHI', linktype, 0, snaplen) + return self.block(BLOCK_IDB, body + self.options(options or [])) + + def epb(self, interface: 'int', timestamp: 'int', data: 'bytes', + options: 'Optional[list[tuple[int, bytes]]]' = None, + original_len: 'Optional[int]' = None) -> 'bytes': + """Enhanced Packet Block. + + Args: + interface: Index of the interface description block it belongs to. + timestamp: 64-bit timestamp, split high word first. + data: Captured packet bytes. + options: ``epb_*`` options, if any. + original_len: On-the-wire length, when the packet was truncated. + + """ + body = self.pack('IIIII', interface, timestamp >> 32, timestamp & 0xFFFF_FFFF, + len(data), len(data) if original_len is None else original_len) + return self.block(BLOCK_EPB, body + self.pad(data) + self.options(options or [])) + + def spb(self, data: 'bytes', original_len: 'Optional[int]' = None) -> 'bytes': + """Simple Packet Block.""" + body = self.pack('I', len(data) if original_len is None else original_len) + return self.block(BLOCK_SPB, body + self.pad(data)) + + def nrb(self, records: 'list[tuple[int, bytes]]', + options: 'Optional[list[tuple[int, bytes]]]' = None) -> 'bytes': + """Name Resolution Block. + + The record list is terminated by an ``nrb_record_end`` record, which the + specification requires even when the option area is empty. + + """ + body = b''.join(self.pack('HH', kind, len(value)) + self.pad(value) + for kind, value in records) + body += self.pack('HH', NRB_RECORD_END, 0) + return self.block(BLOCK_NRB, body + self.options(options or [])) + + def isb(self, interface: 'int', timestamp: 'int', + options: 'Optional[list[tuple[int, bytes]]]' = None) -> 'bytes': + """Interface Statistics Block.""" + body = self.pack('III', interface, timestamp >> 32, timestamp & 0xFFFF_FFFF) + return self.block(BLOCK_ISB, body + self.options(options or [])) + + def dsb(self, secrets_type: 'int', secrets: 'bytes', + options: 'Optional[list[tuple[int, bytes]]]' = None) -> 'bytes': + """Decryption Secrets Block.""" + body = self.pack('II', secrets_type, len(secrets)) + self.pad(secrets) + return self.block(BLOCK_DSB, body + self.options(options or [])) + + def journal(self, entry: 'bytes') -> 'bytes': + """:manpage:`systemd(1)` Journal Export Block. + + ``entry`` is padded out to a 4-octet boundary by the caller rather than + here -- see :func:`_journal_entry` for why that distinction matters. + + """ + return self.block(BLOCK_JOURNAL, entry) + + +def _timestamp(offset_us: 'int') -> 'int': + """Microsecond timestamp ``offset_us`` after :data:`BASE_SECONDS`.""" + return BASE_SECONDS * 1_000_000 + offset_us + + +def _read_epb_packets(path: 'pathlib.Path') -> 'list[bytes]': + """Extract the enhanced packet block payloads from a PCAP-NG file. + + A deliberately minimal reader: it walks the block chain, tracks the byte + order declared by each section header, and returns the captured bytes of + every enhanced packet block. It exists so the synthesised fixtures can carry + real DHCP traffic instead of hand-rolled filler. + + Args: + path: PCAP-NG file to read. + + Returns: + Captured packet payloads, in file order. + + Raises: + RuntimeError: If ``path`` is missing or is not a PCAP-NG file. + + """ + if not path.is_file(): + raise RuntimeError(f'{path} is missing; it is a committed fixture and ' + 'the synthesised captures are derived from it') + + data = path.read_bytes() + packets = [] # type: list[bytes] + endian, offset = '<', 0 + + while offset + 12 <= len(data): + block_type, = struct.unpack_from(endian + 'I', data, offset) + if block_type == BLOCK_SHB: + magic, = struct.unpack_from('' + total, = struct.unpack_from(endian + 'I', data, offset + 4) + if total < 12 or offset + total > len(data): + raise RuntimeError(f'{path}: bad block length {total} at offset {offset}') + + if block_type == BLOCK_EPB: + captured, = struct.unpack_from(endian + 'I', data, offset + 20) + packets.append(data[offset + 28:offset + 28 + captured]) + offset += total + + if not packets: + raise RuntimeError(f'{path}: no enhanced packet blocks found') + return packets + + +def _journal_entry(message: 'str', realtime_us: 'int') -> 'bytes': + """Build a :manpage:`systemd(1)` journal export entry of 4-octet length. + + The block body is padded to a 4-octet boundary like every other PCAP-NG + block, but pcapkit hands the padding to its journal-entry parser along with + the entry itself, and the parser then reads the NULs as the start of a + binary field and raises ``struct.error: unpack requires a buffer of 8 + bytes``. Choosing a message whose entry is already aligned means no padding + is added, which the specification allows and which keeps the fixture + parseable. The underlying defect is + ``pcapkit/protocols/schema/misc/pcapng.py:1376`` splitting ``self.entry`` + without first stripping the block padding. + + """ + while True: + entry = (f'__REALTIME_TIMESTAMP={realtime_us}\n' + f'_TRANSPORT=journal\n' + f'PRIORITY=6\n' + f'MESSAGE={message}\n' + f'\n').encode('utf-8') + if len(entry) % 4 == 0: + return entry + message += '.' + + +def _tls_key_log() -> 'bytes': + """A synthetic, deterministic TLS key log for the decryption secrets block. + + The values are counted-up filler, not captured key material -- nothing here + decrypts anything. + + """ + client_random = bytes(range(0x20, 0x40)).hex() + secret = bytes(range(0x40, 0x70)).hex() + return f'CLIENT_RANDOM {client_random} {secret}\n'.encode('ascii') + + +def _timestamp_options(writer: '_Blocks', + resolution: 'int' = 6) -> 'list[tuple[int, bytes]]': + """The three timestamp options every synthesised interface carries. + + pcapkit looks ``if_tsresol``, ``if_tzone`` and ``if_tsoffset`` up on the + owning interface for each packet block and logs a ``MissingKeyError`` for + each one that is absent -- harmless, but it buries the interesting output + under three lines per packet. Declaring them explicitly keeps the fixtures + quiet as well as complete. + + Args: + writer: Writer whose byte order the option values are packed in. + resolution: ``if_tsresol`` exponent, 6 for microseconds, 9 for + nanoseconds. + + """ + return [ + (IF_TSRESOL, bytes([resolution])), + (IF_TZONE, writer.pack('i', 0)), + (IF_TSOFFSET, writer.pack('q', 0)), + ] + + +def _interface_profile(writer: '_Blocks', name: 'str', description: 'str', + address: 'str', mac: 'bytes', + resolution: 'int' = 6) -> 'list[tuple[int, bytes]]': + """The full set of ``if_*`` options pcapkit can parse for one interface. + + ``if_IPv6addr`` is the one omission, and it is omitted because pcapkit + raises on it rather than because it does not belong here -- see the module + docstring. + + Args: + writer: Writer whose byte order the option values are packed in. + name: ``if_name`` value. + description: ``if_description`` value. + address: Dotted-quad IPv4 address for ``if_IPv4addr``. + mac: Six-octet hardware address for ``if_MACaddr``. + resolution: ``if_tsresol`` exponent. + + """ + octets = bytes(int(part) for part in address.split('.')) + return [ + (OPT_COMMENT, f'synthetic interface {name}'.encode('utf-8')), + (IF_NAME, name.encode('utf-8')), + (IF_DESCRIPTION, description.encode('utf-8')), + (IF_IPV4ADDR, octets + bytes([255, 255, 255, 0])), + (IF_MACADDR, mac), + (IF_EUIADDR, mac[:3] + b'\xff\xfe' + mac[3:]), + (IF_SPEED, writer.pack('Q', 1_000_000_000)), + (IF_FILTER, b'\x00udp port 67 or udp port 68'), + (IF_OS, b'synthetic capture host'), + (IF_FCSLEN, b'\x04'), + (IF_HARDWARE, b'PyPCAPKit synthetic adapter'), + (IF_TXSPEED, writer.pack('Q', 1_000_000_000)), + (IF_RXSPEED, writer.pack('Q', 1_000_000_000)), + ] + _timestamp_options(writer, resolution) + + +def _statistics(writer: '_Blocks', interface: 'int', received: 'int', dropped: 'int', + start_us: 'int', end_us: 'int') -> 'list[tuple[int, bytes]]': + """The full set of ``isb_*`` options for one interface statistics block. + + The list is terminated by :meth:`_Blocks.options`, which matters here: + pcapkit sizes an interface statistics block's option area four octets too + generously, and only the explicit ``opt_endofopt`` stops it reading the + trailing block length as an option. + + """ + def split(value: 'int') -> 'bytes': + return writer.pack('II', value >> 32, value & 0xFFFF_FFFF) + + return [ + (OPT_COMMENT, f'statistics for interface {interface}'.encode('utf-8')), + (ISB_STARTTIME, split(start_us)), + (ISB_ENDTIME, split(end_us)), + (ISB_IFRECV, writer.pack('Q', received)), + (ISB_IFDROP, writer.pack('Q', dropped)), + (ISB_FILTERACCEPT, writer.pack('Q', received)), + (ISB_OSDROP, writer.pack('Q', 0)), + (ISB_USRDELIV, writer.pack('Q', received - dropped)), + ] + + +def _packet_options(writer: '_Blocks', index: 'int') -> 'list[tuple[int, bytes]]': + """The full set of ``epb_*`` options for one packet.""" + return [ + (OPT_COMMENT, f'synthetic frame {index}'.encode('utf-8')), + (EPB_FLAGS, writer.pack('I', 0b01 if index % 2 == 0 else 0b10)), + (EPB_HASH, bytes([2]) + writer.pack('I', 0x0BAD_F00D ^ index)), + (EPB_DROPCOUNT, writer.pack('Q', 0)), + (EPB_PACKETID, writer.pack('Q', index)), + (EPB_QUEUE, writer.pack('I', index % 4)), + (EPB_VERDICT, bytes([0]) + writer.pack('Q', 1)), + ] + + +def build_dhcp(endian: 'str') -> 'bytes': + """Build a DHCP capture with a section header of the given byte order. + + Every multi-octet field in the file -- block types, lengths, option codes, + timestamps -- is written in ``endian``, so the fixture genuinely exercises + the byte order it is named for rather than only flipping the magic number. + + Args: + endian: ``'<'`` for little-endian, ``'>'`` for big-endian. + + """ + writer = _Blocks(endian) + packets = _read_epb_packets(DHCP_SOURCE) + + blocks = [writer.shb([ + (OPT_COMMENT, f'DHCP exchange in a {writer.name} section'.encode('utf-8')), + (SHB_HARDWARE, b'synthetic'), + (SHB_OS, b'synthetic capture host'), + (SHB_USERAPPL, b'examples/samples/pcapng.py'), + ])] + blocks.append(writer.idb(LINKTYPE_ETHERNET, 0x0004_0000, [ + (IF_NAME, b'eth0'), + (IF_DESCRIPTION, f'{writer.name} DHCP capture'.encode('utf-8')), + (IF_OS, b'synthetic capture host'), + ] + _timestamp_options(writer))) + + for index, packet in enumerate(packets): + blocks.append(writer.epb(0, _timestamp(index * 1_000_000), packet, [ + (EPB_FLAGS, writer.pack('I', 0b01 if index % 2 == 0 else 0b10)), + ])) + return b''.join(blocks) + + +def build_test() -> 'bytes': + """Build a capture carrying every auxiliary block type pcapkit accepts. + + Two sections, so the section-scoped interface table is exercised too: + + 1. an Ethernet and a raw-IPv4 interface, three packets between them -- + including one truncated -- then a name resolution block, an interface + statistics block, a decryption secrets block and a journal export block; + 2. a second section header with its own interface, which must not inherit + anything from the first, one packet, and the simple packet block (see the + comment at the end of this function for why it lives here). + + """ + writer = _Blocks('<') + packets = _read_epb_packets(DHCP_SOURCE) + + blocks = [writer.shb([ + (OPT_COMMENT, b'auxiliary PCAP-NG block types, section 1 of 2'), + (SHB_HARDWARE, b'synthetic'), + (SHB_OS, b'synthetic capture host'), + (SHB_USERAPPL, b'examples/samples/pcapng.py'), + ])] + blocks.append(writer.idb(LINKTYPE_ETHERNET, 0x0004_0000, + _interface_profile(writer, 'eth0', 'primary Ethernet interface', + '10.0.0.1', b'\x02\x00\x00\x00\x00\x01'))) + blocks.append(writer.idb(LINKTYPE_RAW, 0x0000_FFFF, [ + (IF_NAME, b'raw0'), + (IF_DESCRIPTION, b'raw IPv4 interface, nanosecond timestamps'), + ] + _timestamp_options(writer, resolution=9))) + + # A packet on each interface, the first with the full option set. + blocks.append(writer.epb(0, _timestamp(0), packets[0], _packet_options(writer, 0))) + blocks.append(writer.epb(1, _timestamp(1_000), packets[1][14:])) + + # A truncated packet, so snaplen handling is exercised as well. + blocks.append(writer.epb(0, _timestamp(2_000), packets[2][:32], + original_len=len(packets[2]))) + + # Name resolution. The IPv6 record is included knowing pcapkit mis-sizes + # it -- covering the path is what catches a future crash there. + blocks.append(writer.nrb([ + (NRB_RECORD_IPV4, bytes([10, 0, 0, 1]) + b'gateway.example\x00'), + (NRB_RECORD_IPV4, bytes([10, 0, 0, 2]) + b'client.example\x00alias.example\x00'), + (NRB_RECORD_IPV6, bytes.fromhex('20010db8000000000000000000000001') + + b'v6.example\x00'), + ], [ + (NS_DNSNAME, b'resolver.example'), + (NS_DNSIP4ADDR, bytes([10, 0, 0, 53])), + (NS_DNSIP6ADDR, bytes.fromhex('20010db8000000000000000000000035')), + ])) + + blocks.append(writer.isb(0, _timestamp(3_000), + _statistics(writer, 0, received=4, dropped=1, + start_us=_timestamp(0), + end_us=_timestamp(3_000)))) + blocks.append(writer.dsb(SECRETS_TLS_KEY_LOG, _tls_key_log(), + [(OPT_COMMENT, b'synthetic TLS key log, decrypts nothing')])) + blocks.append(writer.journal(_journal_entry('synthetic journal entry', + _timestamp(4_000)))) + + # Second section, with its own interface table. + blocks.append(writer.shb([ + (OPT_COMMENT, b'auxiliary PCAP-NG block types, section 2 of 2'), + (SHB_USERAPPL, b'examples/samples/pcapng.py'), + ])) + blocks.append(writer.idb(LINKTYPE_ETHERNET, 0x0004_0000, [ + (IF_NAME, b'eth1'), + (IF_DESCRIPTION, b'interface local to the second section'), + ] + _timestamp_options(writer))) + blocks.append(writer.epb(0, _timestamp(5_000), packets[0], + [(OPT_COMMENT, b'first packet of the second section')])) + + # The simple packet block lives here, in the single-interface section, + # rather than in the first section where it would fit the narrative better. + # pcapkit's PCAP-NG engine raises ``FormatError: PCAP-NG: [SPB] invalid + # section with 2 interfaces`` at ``pcapkit/foundation/engines/pcapng.py:218`` + # for any simple packet block in a section declaring more than one + # interface, which the format spec permits: it says only that such a block + # then refers to the first interface description block, not that it is + # invalid. Putting it here keeps the block type covered without shipping a + # fixture the library cannot open. + blocks.append(writer.spb(packets[3])) + return b''.join(blocks) + + +def build_many_interfaces() -> 'bytes': + """Build a capture with eight interface description blocks. + + The offline stand-in for the downloaded Wireshark capture. Packets are dealt + round-robin across the interfaces so every interface index is referenced by + at least one packet block, and each interface gets its own statistics block. + + """ + writer = _Blocks('<') + packets = _read_epb_packets(DHCP_SOURCE) + + interfaces = [ + ('eth0', 'primary Ethernet interface', LINKTYPE_ETHERNET), + ('eth1', 'secondary Ethernet interface', LINKTYPE_ETHERNET), + ('eth2', 'tertiary Ethernet interface', LINKTYPE_ETHERNET), + ('bond0', 'bonded Ethernet interface', LINKTYPE_ETHERNET), + ('br0', 'bridged Ethernet interface', LINKTYPE_ETHERNET), + ('tun0', 'raw IPv4 tunnel interface', LINKTYPE_RAW), + ('tun1', 'second raw IPv4 tunnel interface', LINKTYPE_RAW), + ('lo', 'loopback interface', LINKTYPE_ETHERNET), + ] + + blocks = [writer.shb([ + (OPT_COMMENT, f'{len(interfaces)} interfaces in one section'.encode('utf-8')), + (SHB_HARDWARE, b'synthetic'), + (SHB_OS, b'synthetic capture host'), + (SHB_USERAPPL, b'examples/samples/pcapng.py'), + ])] + for index, (name, description, linktype) in enumerate(interfaces): + blocks.append(writer.idb(linktype, 0x0004_0000, [ + (IF_NAME, name.encode('utf-8')), + (IF_DESCRIPTION, description.encode('utf-8')), + (IF_IPV4ADDR, bytes([10, 0, index, 1]) + bytes([255, 255, 255, 0])), + (IF_MACADDR, b'\x02\x00\x00\x00\x00' + bytes([index + 1])), + (IF_SPEED, writer.pack('Q', 1_000_000_000)), + ] + _timestamp_options(writer))) + + counts = [0] * len(interfaces) + for index in range(len(interfaces) * 3): + interface = index % len(interfaces) + packet = packets[index % len(packets)] + # A raw IPv4 interface carries the frame without its Ethernet header. + if interfaces[interface][2] == LINKTYPE_RAW: + packet = packet[14:] + counts[interface] += 1 + blocks.append(writer.epb(interface, _timestamp(index * 1_000), packet, [ + (EPB_FLAGS, writer.pack('I', 0b01 if index % 2 == 0 else 0b10)), + (EPB_PACKETID, writer.pack('Q', index)), + ])) + + blocks.append(writer.nrb([ + (NRB_RECORD_IPV4, bytes([10, 0, 0, 1]) + b'gateway.example\x00'), + ])) + for index, received in enumerate(counts): + blocks.append(writer.isb(index, _timestamp(100_000), + _statistics(writer, index, received=received, dropped=0, + start_us=_timestamp(0), + end_us=_timestamp(100_000)))) + return b''.join(blocks) + + +def build_profile() -> 'bytes': + """Build a capture profiling two interfaces in depth. + + Where ``test.pcapng`` goes wide over block types, this one goes deep over + the option space that describes an interface and its counters: one interface + carrying every ``if_*`` option pcapkit can parse, a second at nanosecond + timestamp resolution, forty packets between them each with the full + ``epb_*`` option set, and a statistics block per interface carrying every + ``isb_*`` counter. + + """ + writer = _Blocks('<') + packets = _read_epb_packets(DHCP_SOURCE) + + blocks = [writer.shb([ + (OPT_COMMENT, b'interface profile and statistics'), + (SHB_HARDWARE, b'synthetic'), + (SHB_OS, b'synthetic capture host'), + (SHB_USERAPPL, b'examples/samples/pcapng.py'), + ])] + blocks.append(writer.idb(LINKTYPE_ETHERNET, 0x0004_0000, + _interface_profile(writer, 'eth0', + 'fully described Ethernet interface', + '10.0.0.1', b'\x02\x00\x00\x00\x00\x01'))) + blocks.append(writer.idb(LINKTYPE_ETHERNET, 0x0000_FFFF, + _interface_profile(writer, 'eth1', + 'nanosecond resolution interface', + '10.0.1.1', b'\x02\x00\x00\x00\x00\x02', + resolution=9))) + + total = 40 + counts = [0, 0] + for index in range(total): + interface = index % 2 + counts[interface] += 1 + blocks.append(writer.epb(interface, _timestamp(index * 500), + packets[index % len(packets)], + _packet_options(writer, index))) + + for interface, received in enumerate(counts): + blocks.append(writer.isb(interface, _timestamp(total * 500), + _statistics(writer, interface, received=received, + dropped=interface, + start_us=_timestamp(0), + end_us=_timestamp(total * 500)))) + return b''.join(blocks) + + +class _Fixture(NamedTuple): + """One sample fixture, and where its bytes come from.""" + + #: File name, as the test suite spells it. + name: 'str' + #: What PCAP-NG feature the fixture is there to exercise. + feature: 'str' + #: Offline builder, used when there is no URL or the download fails. + build: 'Callable[[], bytes]' + #: Public source URL, if the capture exists upstream. + url: 'Optional[str]' = None + #: SHA-256 of the exact bytes ``url`` is expected to serve. + sha256: 'Optional[str]' = None + + +#: The five fixtures ``tests/protocols/test_pcapng_regression.py`` needs. +FIXTURES = ( + _Fixture( + name='dhcp_big_endian.pcapng', + feature='big-endian section header block', + build=lambda: build_dhcp('>'), + url=WIRESHARK_CAPTURES + 'dhcp_big_endian.pcapng', + sha256='d9706606fc3febb9740897d85818bd06edc76dc7538ea13d8a9131a988376dfb', + ), + _Fixture( + name='dhcp_little_endian.pcapng', + feature='little-endian section header block', + build=lambda: build_dhcp('<'), + ), + _Fixture( + name='many_interfaces.pcapng', + feature='many interface description blocks in one section', + build=build_many_interfaces, + url=WIRESHARK_CAPTURES + 'many_interfaces.pcapng.1', + sha256='1efe50e015468a22a0aed403351fb4fb9bd3a6ce6221430358ad470aad4350be', + ), + _Fixture( + name='test.pcapng', + feature='auxiliary block types, and two sections', + build=build_test, + ), + _Fixture( + name='profile.pcapng', + feature='the if_* and isb_* option space', + build=build_profile, + ), +) + + +def _download(url: 'str') -> 'bytes': + """Fetch ``url`` and return its body.""" + with urllib.request.urlopen(url, timeout=TIMEOUT) as response: # nosec: B310 + return response.read() + + +def _materialise(fixture: '_Fixture', path: 'pathlib.Path') -> 'tuple[bytes, str]': + """Produce one fixture's bytes, and say where they came from. + + A fixture with no URL is built offline. A fixture with a URL is served from + ``path`` when what is already there matches the recorded digest, which makes + repeat runs both idempotent and offline-safe; otherwise it is downloaded and + the digest is checked. + + Args: + fixture: Fixture to produce. + path: Where the fixture will be written, checked for a usable copy. + + Returns: + The fixture's bytes, and a short description of their provenance. + + Raises: + RuntimeError: If a download's digest does not match the recorded one. + + """ + if fixture.url is None: + return fixture.build(), 'synthesised' + + if path.is_file(): + data = path.read_bytes() + if hashlib.sha256(data).hexdigest() == fixture.sha256: + return data, 'downloaded (already present)' + + try: + data = _download(fixture.url) + except (urllib.error.URLError, OSError, ValueError) as exc: + print(f' ! {fixture.url} is unreachable ({exc});') + print(f' ! synthesising a stand-in for {fixture.name} instead -- the ' + 'upstream capture was NOT used') + return fixture.build(), 'synthesised (download unavailable)' + + digest = hashlib.sha256(data).hexdigest() + if digest != fixture.sha256: + raise RuntimeError( + f'{fixture.name}: SHA-256 mismatch for {fixture.url}\n' + f' expected {fixture.sha256}\n' + f' received {digest}\n' + 'The upstream capture has changed, or the download was corrupted. ' + 'Verify the new bytes by hand before updating the recorded digest.' + ) + return data, 'downloaded' + + +class _LogCapture(logging.Handler): + """Collect pcapkit's own log records rather than letting them print. + + pcapkit reports a parse complaint twice -- once through :mod:`warnings` and + once through its ``pcapkit`` logger -- but only the logger call is reliable: + it passes a computed ``stacklevel`` to :func:`warnings.warn` that can point + outside the stack, and the record then never reaches a + :func:`warnings.catch_warnings` recorder. So the logger is what this listens + to. Duplicates are dropped, because one complaint per block in a capture + with sixty-four packets is not sixty-four pieces of information. + + """ + + def __init__(self) -> 'None': + super().__init__(level=logging.WARNING) + self.messages = [] # type: list[str] + + def emit(self, record: 'logging.LogRecord') -> 'None': + """Record ``record``'s message, first occurrence only.""" + message = record.getMessage() + if message not in self.messages: + self.messages.append(message) + + +def _verify(path: 'pathlib.Path') -> 'tuple[str, list[str]]': + """Round-trip one fixture through pcapkit and describe the result. + + Args: + path: Fixture to parse. + + Returns: + A short status string -- a frame count, or the failure -- and the + distinct complaints pcapkit logged while parsing. + + """ + try: + from pcapkit.interface import extract + except ImportError as exc: # pragma: no cover - depends on the environment + return f'not verified ({exc})', [] + + logger = logging.getLogger('pcapkit') + capture = _LogCapture() + level, handlers, propagate = logger.level, logger.handlers, logger.propagate + logger.handlers, logger.propagate = [capture], False + logger.setLevel(logging.WARNING) + try: + extractor = extract(fin=str(path), fout='/tmp/out', format='tree', + store=True, nofile=True) + return f'{extractor.length} frames', capture.messages + except Exception as exc: # pylint: disable=broad-except + return f'FAILED to parse: {type(exc).__name__}: {exc}', capture.messages + finally: + logger.setLevel(level) + logger.handlers, logger.propagate = handlers, propagate + + +def generate(dest: 'pathlib.Path | None' = None) -> 'list[pathlib.Path]': + """Write the ``.pcapng`` sample fixtures into ``dest``. + + Idempotent: a fixture already on disk with the right bytes is left alone, + and a downloaded fixture already present with the right digest is not + fetched again. Every fixture written is parsed back with pcapkit and the + result reported, so a fixture that the library cannot read is loud rather + than silent. + + Args: + dest: Destination directory; defaults to ``/sample``. + + Returns: + The paths written, in fixture order. + + Raises: + RuntimeError: If a download's digest does not match, or if any fixture + fails to parse back through pcapkit. + + """ + root = SAMPLE_DIR if dest is None else pathlib.Path(dest) + root.mkdir(parents=True, exist_ok=True) + + print(f'writing PCAP-NG fixtures into {root}') + written = [] # type: list[pathlib.Path] + broken = [] # type: list[str] + + for fixture in FIXTURES: + path = root / fixture.name + data, origin = _materialise(fixture, path) + + if path.is_file() and path.read_bytes() == data: + action = 'unchanged' + else: + path.write_bytes(data) + action = 'written' + + status, complaints = _verify(path) + if status.startswith('FAILED'): + broken.append(f'{fixture.name}: {status}') + + print(f' [{origin}] {fixture.name}') + print(f' {len(data)} bytes, {action}; {status}') + print(f' exercises {fixture.feature}') + for complaint in complaints: + print(f' note: {complaint}') + written.append(path) + + if broken: + raise RuntimeError('pcapkit could not parse the following fixture(s):\n ' + + '\n '.join(broken)) + return written + + +if __name__ == '__main__': + generate(pathlib.Path(sys.argv[1]) if len(sys.argv) > 1 else None) diff --git a/pcapkit/foundation/engines/__init__.py b/pcapkit/foundation/engines/__init__.py index f4c45348d6..760c87f3bf 100644 --- a/pcapkit/foundation/engines/__init__.py +++ b/pcapkit/foundation/engines/__init__.py @@ -10,10 +10,6 @@ :mod:`PyShark `, :mod:`DPKT ` 3rd party engine support. -.. todo:: - - Implement support for `PCAP-NG`_ file format. - .. _PCAPNG: https://wiki.wireshark.org/Development/PcapNg """ diff --git a/setup.py b/setup.py index b86edbf832..e60c56fa9b 100644 --- a/setup.py +++ b/setup.py @@ -30,7 +30,8 @@ def get_long_description() -> "str": """Extract description from README.rst, for PyPI's usage.""" - with open("README.rst", encoding="utf-8") as file: + readme = os.path.join(os.path.dirname(os.path.abspath(__file__)), "README.rst") + with open(readme, encoding="utf-8") as file: long_description = file.read() return long_description diff --git a/tests/_support.py b/tests/_support.py index 2fe1a4d695..371bd0338f 100644 --- a/tests/_support.py +++ b/tests/_support.py @@ -7,6 +7,44 @@ from typing import Iterable ROOT = pathlib.Path(__file__).resolve().parents[1] +SAMPLE_ROOT = ROOT / 'examples' / 'sample' +REGENERATE_SAMPLES_CMD = 'python examples/samples/make_samples.py' + + +def sample_path(name: str) -> str: + """Resolve a sample capture file name to its absolute path. + + The sample captures live in :file:`examples/sample/` under the repository + root. Tests go through this helper rather than spelling that directory out, + so that the location is recorded in exactly one place and so that the suite + does not depend on the working directory :program:`pytest` was invoked from. + + Args: + name: Bare file name of the capture, e.g. ``'arp.pcap'`` -- not a path, + and in particular not ``'sample/arp.pcap'``. + + Returns: + Absolute path to the capture as a :obj:`str`, ready to be handed to + :func:`pcapkit.interface.extract` as its ``fin`` argument. + :obj:`str` rather than :class:`pathlib.Path` is deliberate: + :meth:`Extractor.make_name ` + branches on ``isinstance(fin, str)`` and treats anything else as an + already-open binary IO object. + + Raises: + FileNotFoundError: If the capture is not present. Most of the samples + are generated rather than committed to the repository, so a fresh + clone has to build them first. + + """ + path = SAMPLE_ROOT / name + if not path.is_file(): + raise FileNotFoundError( + f'sample capture {name!r} not found at {path} -- most of the sample ' + f'captures are generated, not committed; regenerate them by running ' + f'{REGENERATE_SAMPLES_CMD!r} from {ROOT}' + ) + return str(path) def ensure_package(name: str, path: pathlib.Path) -> types.ModuleType: diff --git a/tests/integration/test_engine_runtime.py b/tests/integration/test_engine_runtime.py index fc4b1423b2..47a69cff32 100644 --- a/tests/integration/test_engine_runtime.py +++ b/tests/integration/test_engine_runtime.py @@ -5,7 +5,7 @@ import sys import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) HAS_DPKT = importlib.util.find_spec('dpkt') is not None @@ -21,7 +21,7 @@ def setUp(self) -> None: def test_default_engine_exposes_native_frame_objects(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/in.pcap', fout='/tmp/out', format='tree', store=True, nofile=True, engine='default') + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True, engine='default') self.addCleanup(close_extractor, extractor) self.assertEqual(extractor.length, 6) @@ -33,7 +33,7 @@ def test_dpkt_engine_returns_dpkt_packets_and_toolkit_chain(self) -> None: from pcapkit.interface import extract from pcapkit.toolkit.dpkt import packet2chain - extractor = extract(fin='sample/in.pcap', fout='/tmp/out', format='tree', store=True, nofile=True, engine='dpkt') + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True, engine='dpkt') self.addCleanup(close_extractor, extractor) frame = extractor.frame[0] @@ -45,7 +45,7 @@ def test_dpkt_engine_returns_dpkt_packets_and_toolkit_chain(self) -> None: def test_scapy_engine_returns_scapy_packets(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/in.pcap', fout='/tmp/out', format='tree', store=True, nofile=True, engine='scapy') + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True, engine='scapy') self.addCleanup(close_extractor, extractor) frame = extractor.frame[0] @@ -63,9 +63,9 @@ def test_pyshark_engine_currently_breaks_on_modern_asyncio(self) -> None: if sys.version_info >= (3, 14): with self.assertRaises(AttributeError): - extract(fin='sample/in.pcap', fout='/tmp/out', format='tree', store=True, nofile=True, engine='pyshark') + extract(fin=sample_path('in.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True, engine='pyshark') else: - extractor = extract(fin='sample/in.pcap', fout='/tmp/out', format='tree', store=True, nofile=True, engine='pyshark') + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True, engine='pyshark') self.addCleanup(close_extractor, extractor) self.assertGreater(extractor.length, 0) diff --git a/tests/integration/test_runtime_extract.py b/tests/integration/test_runtime_extract.py index 2a81820f01..ef49663c57 100644 --- a/tests/integration/test_runtime_extract.py +++ b/tests/integration/test_runtime_extract.py @@ -4,6 +4,8 @@ import os import unittest +from tests._support import sample_path + RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -13,7 +15,7 @@ class RuntimeExtractionTests(unittest.TestCase): def test_extract_reads_small_pcap_sample(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/in.pcap', fout='/tmp/pypcapkit-out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/pypcapkit-out', format='tree', store=True, nofile=True) self.assertEqual(extractor.length, 6) self.assertEqual(len(extractor.frame), 6) @@ -22,7 +24,7 @@ def test_extract_reads_small_pcap_sample(self) -> None: def test_extract_reads_arp_sample(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/arp.pcap', fout='/tmp/pypcapkit-out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('arp.pcap'), fout='/tmp/pypcapkit-out', format='tree', store=True, nofile=True) self.assertEqual(extractor.length, 2) self.assertEqual(len(extractor.frame), 2) @@ -30,7 +32,7 @@ def test_extract_reads_arp_sample(self) -> None: def test_manual_iteration_mode_yields_frames(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/in.pcap', fout='/tmp/pypcapkit-out', format='tree', store=False, nofile=True, auto=False) + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/pypcapkit-out', format='tree', store=False, nofile=True, auto=False) first = next(extractor) second = extractor() diff --git a/tests/interface/test_core.py b/tests/interface/test_core.py index c28dd8d9fb..07784bafa0 100644 --- a/tests/interface/test_core.py +++ b/tests/interface/test_core.py @@ -2,7 +2,7 @@ import unittest -from tests._support import load_module, purge_modules +from tests._support import load_module, purge_modules, sample_path class InterfaceCoreTests(unittest.TestCase): @@ -69,11 +69,11 @@ def test_extract_converts_protocol_type_layer(self) -> None: class DemoProtocol(ProtocolBase): __layer__ = 'internet' - extractor = module.extract(fin='sample/in.pcap', layer=DemoProtocol, store=False) + extractor = module.extract(fin=sample_path('in.pcap'), layer=DemoProtocol, store=False) self.assertEqual(extractor.kwargs['layer'], 'internet') self.assertFalse(extractor.kwargs['store']) - extractor = module.extract(fin='sample/in.pcap', layer='link', store=True) + extractor = module.extract(fin=sample_path('in.pcap'), layer='link', store=True) self.assertEqual(extractor.kwargs['layer'], 'link') self.assertTrue(extractor.kwargs['store']) diff --git a/tests/protocols/application/test_http_runtime.py b/tests/protocols/application/test_http_runtime.py index 051c428ff8..1a6a0683da 100644 --- a/tests/protocols/application/test_http_runtime.py +++ b/tests/protocols/application/test_http_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,12 +17,12 @@ def setUp(self) -> None: def _extract(self, sample: str): from pcapkit.interface import extract - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=True, nofile=True) self.addCleanup(close_extractor, extractor) return extractor def test_http_request_frame_exposes_receipt_and_headers(self) -> None: - extractor = self._extract('sample/http.pcap') + extractor = self._extract('http.pcap') frame = extractor.frame[114] http = frame.payload.payload.payload.payload receipt = http.info.receipt @@ -41,7 +41,7 @@ def test_http_request_frame_exposes_receipt_and_headers(self) -> None: self.assertIsNone(http.info.body) def test_http_response_frame_exposes_status_headers_and_body(self) -> None: - extractor = self._extract('sample/http.pcap') + extractor = self._extract('http.pcap') frame = extractor.frame[117] http = frame.payload.payload.payload.payload receipt = http.info.receipt @@ -59,7 +59,7 @@ def test_http_response_frame_exposes_status_headers_and_body(self) -> None: self.assertTrue(http.info.body.startswith(b'\x1f\x8b\x08')) def test_http_request_variants_cover_additional_hosts(self) -> None: - extractor = self._extract('sample/http.pcap') + extractor = self._extract('http.pcap') sports_request = extractor.frame[393].payload.payload.payload.payload beacon_request = extractor.frame[556].payload.payload.payload.payload @@ -77,7 +77,7 @@ def test_http_request_variants_cover_additional_hosts(self) -> None: self.assertIsNone(beacon_request.info.body) def test_http_response_variants_cover_empty_and_small_bodies(self) -> None: - extractor = self._extract('sample/http.pcap') + extractor = self._extract('http.pcap') empty_response = extractor.frame[629].payload.payload.payload.payload small_body_response = extractor.frame[587].payload.payload.payload.payload @@ -94,7 +94,7 @@ def test_http_response_variants_cover_empty_and_small_bodies(self) -> None: self.assertEqual(len(small_body_response.info.body), 35) def test_malformed_http_payload_falls_back_to_raw(self) -> None: - extractor = self._extract('sample/http.pcap') + extractor = self._extract('http.pcap') frame = extractor.frame[1113] tcp = frame.payload.payload.payload diff --git a/tests/protocols/internet/test_ip_runtime.py b/tests/protocols/internet/test_ip_runtime.py index bceafecdbc..c7747fe79f 100644 --- a/tests/protocols/internet/test_ip_runtime.py +++ b/tests/protocols/internet/test_ip_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,12 +17,12 @@ def setUp(self) -> None: def _extract(self, sample: str): from pcapkit.interface import extract - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=True, nofile=True) self.addCleanup(close_extractor, extractor) return extractor def test_ipv4_packet_exposes_addresses_ttl_and_udp_payload(self) -> None: - extractor = self._extract('sample/ipv4.pcap') + extractor = self._extract('ipv4.pcap') frame = extractor.frame[0] ipv4 = frame.payload.payload udp = ipv4.payload @@ -38,7 +38,7 @@ def test_ipv4_packet_exposes_addresses_ttl_and_udp_payload(self) -> None: self.assertEqual(type(udp.payload).__name__, 'Raw') def test_ipv6_icmp_payload_falls_back_to_raw_with_protocol_hint(self) -> None: - extractor = self._extract('sample/ipv6.pcap') + extractor = self._extract('ipv6.pcap') frame = extractor.frame[2] ipv6 = frame.payload.payload raw = ipv6.payload @@ -53,7 +53,7 @@ def test_ipv6_icmp_payload_falls_back_to_raw_with_protocol_hint(self) -> None: self.assertEqual(raw.info.protocol.name, 'IPv6_ICMP') def test_ipv6_udp_packet_exposes_multicast_and_reply_addresses(self) -> None: - extractor = self._extract('sample/stream.pcap') + extractor = self._extract('stream.pcap') frame = extractor.frame[1] ipv6 = frame.payload.payload udp = ipv6.payload @@ -66,7 +66,7 @@ def test_ipv6_udp_packet_exposes_multicast_and_reply_addresses(self) -> None: self.assertEqual(type(udp.payload).__name__, 'Raw') def test_sample_in_tcp_frames_expose_fin_ack_and_null_payload(self) -> None: - extractor = self._extract('sample/in.pcap') + extractor = self._extract('in.pcap') server_fin = extractor.frame[2].payload.payload.payload client_ack = extractor.frame[3].payload.payload.payload @@ -91,7 +91,7 @@ def test_sample_in_tcp_frames_expose_fin_ack_and_null_payload(self) -> None: self.assertEqual(type(client_fin.payload).__name__, 'NoPayload') def test_sample_in_udp_frame_exposes_broadcast_destination_and_raw_payload(self) -> None: - extractor = self._extract('sample/in.pcap') + extractor = self._extract('in.pcap') frame = extractor.frame[5] ipv4 = frame.payload.payload udp = ipv4.payload diff --git a/tests/protocols/internet/test_ipv6_extension_runtime.py b/tests/protocols/internet/test_ipv6_extension_runtime.py index e35bba68d4..5daacb314a 100644 --- a/tests/protocols/internet/test_ipv6_extension_runtime.py +++ b/tests/protocols/internet/test_ipv6_extension_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,12 +17,12 @@ def setUp(self) -> None: def _extract(self, sample: str): from pcapkit.interface import extract - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=True, nofile=True) self.addCleanup(close_extractor, extractor) return extractor def test_ipv6_fragment_chain_records_extension_header_metadata(self) -> None: - extractor = self._extract('sample/ipv6.pcap') + extractor = self._extract('ipv6.pcap') frame = extractor.frame[12] ipv6 = frame.payload.payload frag = list(ipv6.extension_headers.values())[0] @@ -52,7 +52,7 @@ def test_ipv6_fragment_chain_records_extension_header_metadata(self) -> None: def test_ipv6_fragment_extension_forbids_direct_payload_accessors(self) -> None: from pcapkit.utilities.exceptions import UnsupportedCall - extractor = self._extract('sample/ipv6.pcap') + extractor = self._extract('ipv6.pcap') frame = extractor.frame[15] ipv6 = frame.payload.payload frag = list(ipv6.extension_headers.values())[0] diff --git a/tests/protocols/link/test_link_runtime.py b/tests/protocols/link/test_link_runtime.py index 191fbbdcdb..7a09a57040 100644 --- a/tests/protocols/link/test_link_runtime.py +++ b/tests/protocols/link/test_link_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,12 +17,12 @@ def setUp(self) -> None: def _extract(self, sample: str): from pcapkit.interface import extract - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=True, nofile=True) self.addCleanup(close_extractor, extractor) return extractor def test_ethernet_protocol_exposes_addresses_and_next_layer(self) -> None: - extractor = self._extract('sample/arp.pcap') + extractor = self._extract('arp.pcap') ethernet = extractor.frame[0].payload self.assertEqual(type(ethernet).__name__, 'Ethernet') @@ -32,7 +32,7 @@ def test_ethernet_protocol_exposes_addresses_and_next_layer(self) -> None: self.assertEqual(type(ethernet.payload).__name__, 'ARP') def test_arp_protocol_exposes_request_fields_and_nested_raw_payload(self) -> None: - extractor = self._extract('sample/arp.pcap') + extractor = self._extract('arp.pcap') arp = extractor.frame[0].payload.payload self.assertEqual(type(arp).__name__, 'ARP') @@ -45,7 +45,7 @@ def test_arp_protocol_exposes_request_fields_and_nested_raw_payload(self) -> Non self.assertEqual(type(arp.payload).__name__, 'Raw') def test_arp_reply_frame_exposes_reverse_addresses(self) -> None: - extractor = self._extract('sample/arp.pcap') + extractor = self._extract('arp.pcap') arp = extractor.frame[1].payload.payload self.assertEqual(int(arp.info.oper), 2) @@ -55,7 +55,7 @@ def test_arp_reply_frame_exposes_reverse_addresses(self) -> None: self.assertEqual(str(arp.info.tpa), '10.20.30.131') def test_ipv6_chain_from_sample_includes_expected_protocol_names(self) -> None: - extractor = self._extract('sample/in.pcap') + extractor = self._extract('in.pcap') ethernet = extractor.frame[0].payload ipv6 = ethernet.payload raw = ipv6.payload diff --git a/tests/protocols/misc/pcap/test_frame_runtime.py b/tests/protocols/misc/pcap/test_frame_runtime.py index 46a369cd43..c0b9e73303 100644 --- a/tests/protocols/misc/pcap/test_frame_runtime.py +++ b/tests/protocols/misc/pcap/test_frame_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import purge_modules +from tests._support import purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,7 +17,7 @@ def setUp(self) -> None: def test_frame_exposes_expected_public_metadata(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/in.pcap', fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('in.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True) frame = extractor.frame[0] self.assertEqual(frame.name, 'Frame 1') @@ -30,7 +30,7 @@ def test_frame_exposes_expected_public_metadata(self) -> None: def test_frame_packet_and_payload_walk_protocol_stack(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/arp.pcap', fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('arp.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True) frame = extractor.frame[0] self.assertEqual(type(frame.packet).__name__, 'Packet') @@ -41,7 +41,7 @@ def test_frame_packet_and_payload_walk_protocol_stack(self) -> None: def test_frame_info_records_protocol_summary(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/arp.pcap', fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('arp.pcap'), fout='/tmp/out', format='tree', store=True, nofile=True) frame = extractor.frame[0] self.assertEqual(frame.info.protocols, 'Ethernet:ARP:Raw') diff --git a/tests/protocols/test_pcapng_regression.py b/tests/protocols/test_pcapng_regression.py index 77ba950d95..6b2ca4b288 100644 --- a/tests/protocols/test_pcapng_regression.py +++ b/tests/protocols/test_pcapng_regression.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import purge_modules +from tests._support import purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,32 +17,32 @@ def setUp(self) -> None: def test_dhcp_pcapng_extracts_successfully(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/dhcp.pcapng', fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('dhcp.pcapng'), fout='/tmp/out', format='tree', store=True, nofile=True) self.assertGreater(extractor.length, 0) def test_dhcp_big_endian_pcapng_extracts_successfully(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/dhcp_big_endian.pcapng', fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('dhcp_big_endian.pcapng'), fout='/tmp/out', format='tree', store=True, nofile=True) self.assertGreater(extractor.length, 0) def test_dhcp_little_endian_pcapng_extracts_successfully(self) -> None: from pcapkit.interface import extract - extractor = extract(fin='sample/dhcp_little_endian.pcapng', fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path('dhcp_little_endian.pcapng'), fout='/tmp/out', format='tree', store=True, nofile=True) self.assertGreater(extractor.length, 0) def test_additional_pcapng_samples_extract_successfully(self) -> None: from pcapkit.interface import extract samples = ( - 'sample/test.pcapng', - 'sample/many_interfaces.pcapng', - 'sample/profile.pcapng', + 'test.pcapng', + 'many_interfaces.pcapng', + 'profile.pcapng', ) for sample in samples: with self.subTest(sample=sample): - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=True, nofile=True) self.assertGreater(extractor.length, 0) diff --git a/tests/protocols/transport/test_tcp_runtime.py b/tests/protocols/transport/test_tcp_runtime.py index dd1ff8c1c1..48abc6026f 100644 --- a/tests/protocols/transport/test_tcp_runtime.py +++ b/tests/protocols/transport/test_tcp_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,12 +17,12 @@ def setUp(self) -> None: def _extract(self, sample: str): from pcapkit.interface import extract - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=True, nofile=True) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=True, nofile=True) self.addCleanup(close_extractor, extractor) return extractor def test_tcp_ipv4_syn_frame_exposes_expected_option_sequence(self) -> None: - extractor = self._extract('sample/tcp.pcap') + extractor = self._extract('tcp.pcap') frame = extractor.frame[0] tcp = frame.payload.payload.payload options = list(tcp.info.options.items(multi=True)) @@ -57,7 +57,7 @@ def test_tcp_ipv4_syn_frame_exposes_expected_option_sequence(self) -> None: self.assertEqual(type(tcp.payload).__name__, 'NoPayload') def test_tcp_ipv6_frame_keeps_timestamp_only_options(self) -> None: - extractor = self._extract('sample/tcp.pcap') + extractor = self._extract('tcp.pcap') frame = extractor.frame[3] ip = frame.payload.payload tcp = ip.payload @@ -73,7 +73,7 @@ def test_tcp_ipv6_frame_keeps_timestamp_only_options(self) -> None: self.assertEqual(options[2][1].echo, 2559889017) def test_tcp_unregistered_application_payload_falls_back_to_raw(self) -> None: - extractor = self._extract('sample/tcp.pcap') + extractor = self._extract('tcp.pcap') frame = extractor.frame[5] tcp = frame.payload.payload.payload @@ -86,7 +86,7 @@ def test_tcp_unregistered_application_payload_falls_back_to_raw(self) -> None: self.assertIsNone(tcp.payload.info.error) def test_stream_sample_exposes_no_payload_ack_frame(self) -> None: - extractor = self._extract('sample/stream.pcap') + extractor = self._extract('stream.pcap') frame = extractor.frame[3] tcp = frame.payload.payload.payload options = list(tcp.info.options.items(multi=True)) @@ -101,7 +101,7 @@ def test_stream_sample_exposes_no_payload_ack_frame(self) -> None: self.assertEqual(type(tcp.payload).__name__, 'NoPayload') def test_stream_sample_exposes_ipv6_tcp_raw_payload_pair(self) -> None: - extractor = self._extract('sample/stream.pcap') + extractor = self._extract('stream.pcap') client_frame = extractor.frame[2] server_frame = extractor.frame[4] diff --git a/tests/protocols/transport/test_udp_runtime.py b/tests/protocols/transport/test_udp_runtime.py index cc52973a03..a8a429a719 100644 --- a/tests/protocols/transport/test_udp_runtime.py +++ b/tests/protocols/transport/test_udp_runtime.py @@ -3,7 +3,7 @@ import importlib.util import unittest -from tests._support import close_extractor, purge_modules +from tests._support import close_extractor, purge_modules, sample_path RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -17,12 +17,12 @@ def setUp(self) -> None: def _extract(self, sample: str, *, store: bool = True, auto: bool = True): from pcapkit.interface import extract - extractor = extract(fin=sample, fout='/tmp/out', format='tree', store=store, nofile=True, auto=auto) + extractor = extract(fin=sample_path(sample), fout='/tmp/out', format='tree', store=store, nofile=True, auto=auto) self.addCleanup(close_extractor, extractor) return extractor def test_ipv4_udp_frame_exposes_length_checksum_and_raw_payload(self) -> None: - extractor = self._extract('sample/ipv4.pcap', store=False, auto=False) + extractor = self._extract('ipv4.pcap', store=False, auto=False) frame = next(extractor) udp = frame.payload.payload.payload @@ -35,7 +35,7 @@ def test_ipv4_udp_frame_exposes_length_checksum_and_raw_payload(self) -> None: self.assertEqual(type(udp.payload).__name__, 'Raw') def test_ipv6_udp_mdns_frame_exposes_multicast_ports_and_checksum(self) -> None: - extractor = self._extract('sample/stream.pcap') + extractor = self._extract('stream.pcap') frame = extractor.frame[1] udp = frame.payload.payload.payload