tftp.capture

tftp.capture

Seeing TFTP on the wire: packet events, trace hooks, filters, captures.

Three sources of :class:PacketEvent:

• trace= hooks on TFTPClient, TFTPServer and TFTPRelay -- what this library sent and received, with session ids;
• pcap/pcapng captures, from a file or a live tcpdump -w - pipe, read by pktcap (pktcap.read_datagrams) and followed by :class:FlowTracker, or by :func:analyze in one call;
• live capture on Linux (pktcap.sniff).

:func:dissect_tftp is TFTP as a pktcap layer (:class:TFTPLayer), registered by :func:register_tftp_dissector and by nothing else; :func:pktcap_plugin is the hook that lets pktcap load it by name, with the filter keys op, file, block, code and session. :func:follow_transfers gives a transfer's datagrams, which run between ports chosen for it, the layer and the transfer's session.

:func:replay_transfers asks a server the caller names for each transfer a capture holds again.

:func:trace_to turns a pktcap writer (pktcap.PcapWriter, pktcap.PcapngWriter) into a trace hook, so a run is recorded for Wireshark, and :func:combine_hooks joins several trace hooks into the one trace= takes.

Analysis

Bases: NamedTuple

The TFTP in a capture: its events and every transfer seen.

CapturedTransfer(session, time, client, server, request)

One transfer reconstructed from a capture.

:ivar session: id shared with the transfer's :class:PacketEvent objects. :ivar client, server: the request's source, and its destination. :ivar server_tid: the server's transfer address, once it answered. :ivar operation: "read" (RRQ) or "write" (WRQ). :ivar requested, acknowledged: the request's options, the OACK's. :ivar blksize, windowsize: what the transfer ran with. :ivar error: (code, message, sent_by) of an ERROR, or None. :ivar is_complete: the final DATA was seen and acknowledged. :ivar retransmissions: DATA packets seen more than once. :ivar missing_blocks: (first, last) ranges (inclusive) of the blocks never seen, up to the highest one placed; :attr:missing_count counts them.

Source code in src/tftp/capture/_flows.py

def __init__(self, session: str, time: float, client: Endpoint, server: Endpoint, request: Any) -> None:
    self.session = session
    self.started = time
    self.ended = time
    self.client = client
    self.server = server
    self.server_tid: Optional[Endpoint] = None
    self.filename: str = request.filename
    self.mode: str = request.mode
    self.operation = "read" if request.is_read else "write"
    self.requested: Dict[str, str] = dict(request.options)
    self.acknowledged: Dict[str, str] = {}
    self.blksize = DEFAULT_BLKSIZE
    self.windowsize = 1
    self.tsize: Optional[int] = None
    self.error: Optional[Tuple[int, str, str]] = None
    self.is_complete = False
    self.final_block: Optional[int] = None  # logical
    self.packets = 0
    self.retransmissions = 0
    self.request_retransmissions = 0
    self._sizes: Dict[int, int] = {}  # payload octets of each block seen, payloads kept or not
    self._cut: Set[int] = set()  # blocks seen only as a DATA a snap length cut
    self._blocks: Dict[int, bytes] = {}  # the payloads, when they are kept
    self._logical_hi = 0
    self._rollover = 0

missing_blocks property

(first, last) ranges of the logical block numbers never seen whole, up to the highest one seen.

missing_count property

How many blocks :attr:missing_blocks names.

add_data(wire, payload, keep=True, cut=False)

Account for one DATA: always its size, and its payload when keep.

A block further ahead of the highest one seen than a window can be is not placed: it is counted as a packet and nothing else. A DATA a snap length cut is not placed either: its block is missing until a whole one arrives, and its short payload never ends the transfer.

Source code in src/tftp/capture/_flows.py

def add_data(self, wire: int, payload: bytes, keep: bool = True, cut: bool = False) -> None:
    """Account for one DATA: always its size, and its payload when ``keep``.

    A block further ahead of the highest one seen than a window can be is not
    placed: it is counted as a packet and nothing else. A DATA a snap length ``cut`` is
    not placed either: its block is missing until a whole one arrives, and its short
    payload never ends the transfer.
    """
    block = self._logical(wire)
    if block - self._logical_hi > self._reach():
        return
    if cut:
        if block not in self._sizes:
            self._cut.add(block)
            self._logical_hi = max(self._logical_hi, block)
        return
    self._cut.discard(block)
    if block in self._sizes:
        self.retransmissions += 1
    else:
        self._sizes[block] = len(payload)
        if keep:
            self._blocks[block] = payload
    self._logical_hi = max(self._logical_hi, block)
    if len(payload) < self.blksize:
        self.final_block = block

data(decode_netascii=True)

The transferred bytes, in order; a gap ends what can be returned.

Netascii transfers are decoded to local form unless decode_netascii=False. Empty when the tracker did not keep payloads.

Source code in src/tftp/capture/_flows.py

def data(self, decode_netascii: bool = True) -> bytes:
    """The transferred bytes, in order; a gap ends what can be returned.

    Netascii transfers are decoded to local form unless
    ``decode_netascii=False``. Empty when the tracker did not keep payloads.
    """
    out = bytearray()
    block = 1
    while block in self._blocks:
        out += self._blocks[block]
        block += 1
    raw = bytes(out)
    return netascii_decode(raw) if decode_netascii and self.mode == "netascii" else raw

write_to(directory)

Write the transferred bytes into directory and return the path, or None if there is nothing.

The file is named <session>-<name>, with .partial added when the transfer is not complete or has gaps. <name> is the last component of the requested name with every character outside A-Za-z0-9_.- replaced by _ and at most 100 octets long, so a capture cannot name a file outside directory. The directory is created when it does not exist. Without kept payloads (keep_payloads=False) there is nothing to write.

Source code in src/tftp/capture/_flows.py

def write_to(self, directory: "Union[str, os.PathLike[str]]") -> Optional[str]:
    """Write the transferred bytes into ``directory`` and return the path, or ``None`` if there is nothing.

    The file is named ``<session>-<name>``, with ``.partial`` added when the
    transfer is not complete or has gaps. ``<name>`` is the last component of
    the requested name with every character outside ``A-Za-z0-9_.-`` replaced
    by ``_`` and at most 100 octets long, so a capture cannot name a file
    outside ``directory``. The directory is created when it does not exist.
    Without kept payloads (``keep_payloads=False``) there is nothing to write.
    """
    data = self.data()
    if not data and not self.is_complete:
        return None
    base = self.filename.replace("\\", "/").rsplit("/", 1)[-1]
    name = _UNSAFE_NAME.sub("_", base)[:_NAME_LENGTH] or "file"
    partial = "" if self.is_complete and not self.missing_blocks else ".partial"
    os.makedirs(directory, exist_ok=True)
    path = os.path.join(directory, "%s-%s%s" % (self.session, name, partial))
    with open(path, "wb") as handle:
        handle.write(data)
    return path

DatagramLike

Bases: Protocol

What :meth:FlowTracker.feed reads of a datagram.

Required: time, source, destination and payload.

Optional member, read once per datagram when present:

• truncated -- True when payload is shorter than the datagram was because a snap length cut the capture. Such a datagram is counted as a packet of its transfer but never read as a DATA: a short payload is not the final block, and its block is reported missing. Without the member a datagram is whole.

pktcap.CapturedDatagram has all five.

DatagramWriter

Bases: Protocol

What :func:trace_to needs of a writer: pktcap.PcapWriter and pktcap.PcapngWriter have it.

write(time, source, destination, payload)

Append one UDP datagram: seconds since the epoch, both (host, port) addresses, its octets.

Source code in src/tftp/capture/_hook.py

def write(
    self, time: float, source: Tuple[Any, ...], destination: Tuple[Any, ...], payload: bytes
) -> None:
    """Append one UDP datagram: seconds since the epoch, both ``(host, port)`` addresses, its octets."""
    ...

FlowTracker(ports=(69,), *, keep_payloads=True, on_complete=None, max_tracked=1024)

Assigns datagrams to transfers and reconstructs them.

:param ports: request ports (where RRQ/WRQ are sent). :param keep_payloads: keep DATA payloads (for :meth:CapturedTransfer.data). Turn off for long captures where only metadata matters: sizes, retransmissions and missing blocks are counted either way. :param on_complete: on_complete(transfer) for each transfer the tracker stops following (see max_tracked). :param max_tracked: at most this many transfers are followed; None follows every one, as a whole file's analysis wants. When a request would make one more, the transfer that is finished (complete or ended by an ERROR) and quietest goes first, else the quietest one: it is handed to on_complete and dropped, so a live capture holds a bounded number of transfers however long it runs. A datagram of a dropped transfer is no longer attributed to it. :raises ValueError: max_tracked below 1.

Source code in src/tftp/capture/_flows.py

def __init__(
    self,
    ports: Iterable[int] = (69,),
    *,
    keep_payloads: bool = True,
    on_complete: "Optional[Callable[[CapturedTransfer], Any]]" = None,
    max_tracked: Optional[int] = 1024,
) -> None:
    if max_tracked is not None and max_tracked < 1:
        raise ValueError("max_tracked is at least 1, or None for no bound")
    self.ports = frozenset(ports)
    self.keep_payloads = keep_payloads
    self.on_complete = on_complete
    self.max_tracked = max_tracked
    #: The transfers being followed, oldest request first.
    self.transfers: List[CapturedTransfer] = []
    self._by_client: Dict[Endpoint, CapturedTransfer] = {}

feed(datagram)

Account for one datagram; its event if it is TFTP, else None.

Source code in src/tftp/capture/_flows.py

def feed(self, datagram: DatagramLike) -> Optional[PacketEvent]:
    """Account for one datagram; its event if it is TFTP, else ``None``."""
    time, payload = datagram.time, datagram.payload
    cut = bool(getattr(datagram, "truncated", False))
    source, destination = _plain(datagram.source), _plain(datagram.destination)
    transfer = None
    if destination[1] in self.ports and len(payload) >= 2 and payload[0] == 0 and payload[1] in (1, 2):
        try:
            request = RequestPacket.decode(payload)
        except TFTPDecodeError:
            request = None
        if request is not None:
            current = self._by_client.get(source)
            if (
                current is not None
                and current.server_tid is None
                and current.filename == request.filename
            ):
                current.request_retransmissions += 1  # the same request again
                transfer = current
            else:
                transfer = CapturedTransfer(new_session_id("c"), time, source, destination, request)
                self._make_room()
                self.transfers.append(transfer)
                self._by_client[source] = transfer
    else:
        transfer = self._match(time, source, destination)
    if transfer is None:
        if destination[1] in self.ports or source[1] in self.ports:
            return PacketEvent(time, "seen", destination, source, payload, "capture")
        return None
    transfer.packets += 1
    transfer.ended = time
    try:
        self._observe(transfer, source, payload, cut)
    except Exception as exc:  # a packet this reading cannot take fails its transfer, not the capture
        if transfer.error is None:
            transfer.error = (0, "unreadable packet: %s" % type(exc).__name__, "capture")
    return PacketEvent(time, "seen", destination, source, payload, "capture", transfer.session)

PacketEvent

Bases: NamedTuple

A datagram seen by a client, server, relay or capture.

:ivar time: wall-clock seconds (time.time()). :ivar direction: "in" (received) or "out" (sent), from role's point of view; "seen" for a passive capture. :ivar local: the observer's address for this datagram (for a capture, the destination). :ivar remote: the other end (for a capture, the source). :ivar data: the datagram itself. :ivar role: "client", "server", "relay" or "capture". :ivar session: id correlating one transfer's events, or None for datagrams outside any transfer (requests that were refused...). :ivar leg: for a relay, "client" or "upstream": which side of it.

destination property

Who it was sent to.

source property

Who sent it.

__str__()

A human line: 12:00:00.123456 [t3] 10.0.0.5:2000 > 10.0.0.1:69 RRQ 'f' octet.

Source code in src/tftp/capture/_events.py

def __str__(self) -> str:
    """A human line: ``12:00:00.123456 [t3] 10.0.0.5:2000 > 10.0.0.1:69 RRQ 'f' octet``."""
    try:
        stamp = datetime.datetime.fromtimestamp(self.time).strftime("%H:%M:%S.%f")
    except (OverflowError, OSError, ValueError):  # a time the platform cannot convert
        stamp = "%.6f" % self.time
    tags = " ".join(t for t in (self.session and "[%s]" % self.session, self.leg) if t)
    return "%s %s%s > %s %s" % (
        stamp,
        tags + " " if tags else "",
        _endpoint(self.source),
        _endpoint(self.destination),
        self.summary,
    )

decode()

The parsed packet; raises :class:TFTPDecodeError.

Source code in src/tftp/capture/_events.py

def decode(self) -> Any:
    """The parsed packet; raises :class:`TFTPDecodeError`."""
    return decode(self.data)

to_dict(payload=False)

JSON-ready metadata; DATA payloads only with payload=True (hex).

Source code in src/tftp/capture/_events.py

def to_dict(self, payload: bool = False) -> Dict[str, Any]:
    """JSON-ready metadata; DATA payloads only with ``payload=True`` (hex)."""
    record: Dict[str, Any] = {
        "time": self.time,
        "direction": self.direction,
        "role": self.role,
        "session": self.session,
        "leg": self.leg,
        "source": _endpoint(self.source),
        "destination": _endpoint(self.destination),
        "opcode": self.opcode_name,
        "length": len(self.data),
        "summary": self.summary,
    }
    if self.block is not None:
        record["block"] = self.block
    try:
        packet = self.decode()
    except TFTPDecodeError:
        packet = None
    if packet is not None:
        for field in ("filename", "mode", "options", "code", "message"):
            if hasattr(packet, field):
                value = getattr(packet, field)
                record[field] = (
                    int(value) if field == "code" else dict(value) if field == "options" else value
                )
    if payload and self.opcode == TFTPOpcode.DATA:
        record["payload"] = self.data[4:].hex()
    return record

ReplayedTransfers

Bases: NamedTuple

What :func:replay_transfers did: the transfers it ran, and how many it did not.

TFTPLayer

Bases: NamedTuple

The TFTP packet a datagram carries, as plain values.

opcode is the packet's name ("RRQ", "WRQ", "DATA", "ACK", "ERROR", "OACK"). Every other field is None where that kind of packet has none: block (DATA, ACK), filename and mode (RRQ, WRQ, mode lower-cased), options (RRQ, WRQ, OACK: (name, value) pairs, names lower-cased, in the order sent), code and message (ERROR). session is the id of the transfer the datagram belongs to, None for a layer read from the octets alone: :func:follow_transfers sets it. Equal to another TFTPLayer with the same fields and to nothing else.

summary()

One line saying what the packet is, for pktcap's text output: [c1] RRQ 'boot.efi' octet.

The transfer's id leads when it is known. The file name, the mode, the options and an ERROR's message are the peer's text, shown as it came: pktcap escapes the line. A field a hand-built layer lacks reads as None; nothing here raises.

Source code in src/tftp/capture/_dissector.py

def summary(self) -> str:
    """One line saying what the packet is, for pktcap's ``text`` output: ``[c1] RRQ 'boot.efi' octet``.

    The transfer's id leads when it is known. The file name, the mode, the options and an ERROR's
    message are the peer's text, shown as it came: pktcap escapes the line. A field a hand-built
    layer lacks reads as ``None``; nothing here raises.
    """
    if self.opcode in ("RRQ", "WRQ"):
        text = "%s %s %s%s" % (self.opcode, _quote(self.filename), self.mode, _options(self.options))
    elif self.opcode in ("DATA", "ACK"):
        text = "%s %s" % (self.opcode, self.block)
    elif self.opcode == "ERROR":
        text = "ERROR %s %s" % (self.code, _quote(self.message))
    elif self.opcode == "OACK":
        text = "OACK" + _options(self.options)
    else:
        text = str(self.opcode)
    return "[%s] %s" % (self.session, text) if self.session else text

analyze(source, *, ports=(69,), keep_payloads=True)

Read a whole capture (path, stream, or datagrams) and reconstruct its transfers.

A path or a stream is read by pktcap.read_datagrams, which raises pktcap.CaptureFormatError for a file that is not a capture. Every transfer is kept: the result holds what the capture holds.

Source code in src/tftp/capture/_analysis.py

def analyze(
    source: Union[str, "os.PathLike[str]", BinaryIO, Iterable[DatagramLike]],
    *,
    ports: Iterable[int] = (69,),
    keep_payloads: bool = True,
) -> Analysis:
    """Read a whole capture (path, stream, or datagrams) and reconstruct its transfers.

    A path or a stream is read by ``pktcap.read_datagrams``, which raises
    ``pktcap.CaptureFormatError`` for a file that is not a capture. Every
    transfer is kept: the result holds what the capture holds.
    """
    if isinstance(source, (str, os.PathLike)) or hasattr(source, "read"):
        require_pktcap()
        from pktcap import read_datagrams

        datagrams: Iterable[DatagramLike] = read_datagrams(
            cast("Union[str, os.PathLike[str], BinaryIO]", source)
        )
    else:
        datagrams = source
    tracker = FlowTracker(ports, keep_payloads=keep_payloads, max_tracked=None)
    return Analysis(list(tracker.feed_all(datagrams)), tracker.transfers)

combine_hooks(*hooks)

One trace hook that calls each of hooks in order; None entries are skipped.

Returns None when no hook is left, and the hook itself when one is. A hook that raises does not keep the others from seeing the event: the first exception is raised once all have been called, for the guard around the combined hook to count.

Source code in src/tftp/capture/_hook.py

def combine_hooks(*hooks: Optional[Callable[[Any], Any]]) -> Optional[Callable[[Any], None]]:
    """One trace hook that calls each of ``hooks`` in order; ``None`` entries are skipped.

    Returns ``None`` when no hook is left, and the hook itself when one is. A
    hook that raises does not keep the others from seeing the event: the first
    exception is raised once all have been called, for the guard around the
    combined hook to count.
    """
    present = [hook for hook in hooks if hook is not None]
    if not present:
        return None
    if len(present) == 1:
        return present[0]

    def combined(event: Any) -> None:
        failure: Optional[Exception] = None
        for hook in present:
            try:
                hook(event)
            except Exception as exc:
                failure = failure or exc
        if failure is not None:
            raise failure

    return combined

dissect_tftp(data)

Read the TFTP packet data holds: its :class:TFTPLayer, and a DATA's octets as the payload.

The payload of every other packet is empty, and nothing follows a TFTP packet. Raises pktcap.DissectError (a ValueError) for octets that are not a TFTP packet; its text holds none of them. Follows the contract pktcap.check_dissector checks.

Source code in src/tftp/capture/_dissector.py

def dissect_tftp(data: bytes) -> "pktcap.Dissected":
    """Read the TFTP packet ``data`` holds: its :class:`TFTPLayer`, and a DATA's octets as the payload.

    The payload of every other packet is empty, and nothing follows a TFTP packet. Raises
    ``pktcap.DissectError`` (a ``ValueError``) for octets that are not a TFTP packet; its text
    holds none of them. Follows the contract ``pktcap.check_dissector`` checks.
    """
    require_pktcap()
    import pktcap

    try:
        packet = decode(data)
    except TFTPDecodeError:
        raise pktcap.DissectError("not a TFTP packet") from None
    payload = b""
    if isinstance(packet, RequestPacket):
        layer = TFTPLayer(
            packet.opcode.name,
            filename=packet.filename,
            mode=packet.mode,
            options=tuple(packet.options.items()),
        )
    elif isinstance(packet, DataPacket):
        layer, payload = TFTPLayer("DATA", block=packet.block), packet.data
    elif isinstance(packet, AckPacket):
        layer = TFTPLayer("ACK", block=packet.block)
    elif isinstance(packet, ErrorPacket):
        layer = TFTPLayer("ERROR", code=int(packet.code), message=packet.message)
    else:
        layer = TFTPLayer("OACK", options=tuple(packet.options.items()))
    return pktcap.Dissected(layer, payload)

follow_transfers(frames, tracker)

frames with a :class:TFTPLayer on each datagram tracker attributes to a transfer.

pktcap dissects TFTP by port, so only a datagram to or from the request port has a layer; a transfer's DATA, ACK and OACK run between ports chosen for it. Every UDP datagram is fed to tracker, and one it attributes to a transfer is yielded with its layer, whose session is the transfer's: a layer the dissector already made is replaced, otherwise the layer is added after the UDP layer and the octets that follow it (a DATA's) become the frame's payload. A frame with no UDP datagram, one the tracker does not attribute, one a snap length cut (the tracker counts it) and one whose octets are not a TFTP packet are yielded as they are. The result is lazy and keeps the order; nothing is read from frames before it is iterated, and closing the result closes frames when it can be closed. ImportError when pktcap is not installed.

Source code in src/tftp/capture/_follow.py

def follow_transfers(
    frames: "Iterable[pktcap.DissectedFrame]", tracker: FlowTracker
) -> "Iterator[pktcap.DissectedFrame]":
    """``frames`` with a :class:`TFTPLayer` on each datagram ``tracker`` attributes to a transfer.

    pktcap dissects TFTP by port, so only a datagram to or from the request port has a layer; a
    transfer's DATA, ACK and OACK run between ports chosen for it. Every UDP datagram is fed to
    ``tracker``, and one it attributes to a transfer is yielded with its layer, whose ``session`` is
    the transfer's: a layer the dissector already made is replaced, otherwise the layer is added
    after the UDP layer and the octets that follow it (a DATA's) become the frame's payload. A frame
    with no UDP datagram, one the tracker does not attribute, one a snap length cut (the tracker
    counts it) and one whose octets are not a TFTP packet are yielded as they are. The result is
    lazy and keeps the order; nothing is read from ``frames`` before it is iterated, and closing the
    result closes ``frames`` when it can be closed. ``ImportError`` when pktcap is not installed.
    """
    require_pktcap()
    return _followed(frames, tracker)

new_session_id(prefix='t')

A short id, unique in this process, for correlating a transfer's events.

Source code in src/tftp/capture/_events.py

def new_session_id(prefix: str = "t") -> str:
    """A short id, unique in this process, for correlating a transfer's events."""
    return "%s%d" % (prefix, next(_session_ids))

pktcap_plugin(registry)

Declare :class:TFTPLayer and its filter keys in registry, then register the dissector.

The keys are op, file, block, code and session (the transfer's id, as :func:follow_transfers sets it, compared exactly), and TFTPLayer fields as tftp.FIELD. A call that raises leaves registry as it found it: ValueError for a layer name or a port that is taken.

Source code in src/tftp/capture/_plugin.py

def pktcap_plugin(registry: "pktcap.DissectorRegistry") -> None:
    """Declare :class:`TFTPLayer` and its filter keys in ``registry``, then register the dissector.

    The keys are ``op``, ``file``, ``block``, ``code`` and ``session`` (the transfer's id, as
    :func:`follow_transfers` sets it, compared exactly), and ``TFTPLayer`` fields as ``tftp.FIELD``.
    A call that raises leaves ``registry`` as it found it: ``ValueError`` for a layer name or a
    port that is taken.
    """
    require_pktcap()
    registry.register_layer(
        TFTPLayer,
        keys={
            "op": _op,
            "file": _file,
            "block": _number("block"),
            "code": _number("code"),
            "session": _session,
        },
    )
    try:
        register_tftp_dissector(registry)
    except Exception:
        registry.unregister_layer("tftp")
        raise

register_tftp_dissector(registry=None, *, ports=(69,))

Have registry (default: pktcap's process-wide one) dissect the datagrams to ports.

Registers :func:dissect_tftp under ("udp", port) for each port. Nothing registers on import and nothing outside this call does. Only the request port is a selector: a transfer's DATA and ACK run between ports chosen per transfer, which :class:FlowTracker follows. ValueError for a port that is not 1 to 65535 or that already has a dissector (pktcap's rule), in which case nothing is registered by this call; TypeError for a port that is not an int.

Source code in src/tftp/capture/_dissector.py

def register_tftp_dissector(
    registry: Optional["pktcap.DissectorRegistry"] = None, *, ports: Iterable[int] = (69,)
) -> None:
    """Have ``registry`` (default: pktcap's process-wide one) dissect the datagrams to ``ports``.

    Registers :func:`dissect_tftp` under ``("udp", port)`` for each port. Nothing registers on
    import and nothing outside this call does. Only the request port is a selector: a transfer's
    DATA and ACK run between ports chosen per transfer, which :class:`FlowTracker` follows.
    ``ValueError`` for a port that is not 1 to 65535 or that already has a dissector (pktcap's
    rule), in which case nothing is registered by this call; ``TypeError`` for a port that is
    not an ``int``.
    """
    require_pktcap()
    import pktcap

    wanted: List[Any] = list(ports)
    for port in wanted:
        if isinstance(port, bool) or not isinstance(port, int):
            raise TypeError("a port is an int, not %s" % type(port).__name__)
        if not 0 < port < 65536:
            raise ValueError("a port is 1 to 65535, not %d" % port)
    target = registry if registry is not None else pktcap.default_registry()
    done: List[int] = []
    try:
        for port in wanted:
            target.register("udp", port, dissect_tftp)
            done.append(port)
    except ValueError:
        for port in done:
            target.unregister("udp", port)
        raise

replay_transfers(source, host, port=69, *, ports=(69,), writes=False, speed=1.0, max_delay=5.0, limit=None, timeout=1.0, retries=5)

Ask the server at host and port for each transfer the capture source holds again.

source is what :func:analyze takes; ports are its request ports. Transfers run one at a time in the order their requests were seen, each after the wait pktcap.replay_schedule gives (speed, max_delay and limit mean what they mean there: speed=None removes the waits, max_delay is the longest single one, limit ends the replay after that many transfers and the rest are neither run nor counted). Each is requested by a client built from what the capture's client asked for: the mode, the options (timeout is the argument's, not the captured value), and the file name.

Nothing is sent to an address the capture holds: host and port are the only destination. A read is asked for again and its octets are discarded; a write is replayed only with writes=True, and it uploads the octets the capture holds, which overwrites a file on the server. A write the capture holds only partly (a missing block, an ERROR, no final acknowledgement) is never replayed. A transfer that is not replayed, or that the client refuses (a mode it does not send, an option it does not accept), is counted in skipped.

A transfer that fails (an ERROR from the server, a timeout) is a :class:~tftp.TransferResult with error set and the replay goes on. ValueError for an argument out of range; pktcap.CaptureFormatError for a file that is not a capture; OSError when host does not resolve or a socket fails.

Source code in src/tftp/capture/_replay.py

def replay_transfers(
    source: Union[str, "os.PathLike[str]", BinaryIO, Iterable[DatagramLike]],
    host: Any,
    port: int = 69,
    *,
    ports: Iterable[int] = (69,),
    writes: bool = False,
    speed: Optional[float] = 1.0,
    max_delay: float = 5.0,
    limit: Optional[int] = None,
    timeout: float = 1.0,
    retries: int = 5,
) -> ReplayedTransfers:
    """Ask the server at ``host`` and ``port`` for each transfer the capture ``source`` holds again.

    ``source`` is what :func:`analyze` takes; ``ports`` are its request ports. Transfers run one at
    a time in the order their requests were seen, each after the wait ``pktcap.replay_schedule``
    gives (``speed``, ``max_delay`` and ``limit`` mean what they mean there: ``speed=None`` removes
    the waits, ``max_delay`` is the longest single one, ``limit`` ends the replay after that many
    transfers and the rest are neither run nor counted). Each is requested by a client built from
    what the capture's client asked for: the mode, the options (``timeout`` is the argument's, not
    the captured value), and the file name.

    Nothing is sent to an address the capture holds: ``host`` and ``port`` are the only
    destination. **A read is asked for again and its octets are discarded; a write is replayed only
    with ``writes=True``, and it uploads the octets the capture holds, which overwrites a file on
    the server.** A write the capture holds only partly (a missing block, an ERROR, no final
    acknowledgement) is never replayed. A transfer that is not replayed, or that the client refuses
    (a mode it does not send, an option it does not accept), is counted in ``skipped``.

    A transfer that fails (an ERROR from the server, a timeout) is a :class:`~tftp.TransferResult`
    with ``error`` set and the replay goes on. ``ValueError`` for an argument out of range;
    ``pktcap.CaptureFormatError`` for a file that is not a capture; ``OSError`` when ``host`` does
    not resolve or a socket fails.
    """
    require_pktcap()
    import pktcap

    if isinstance(port, bool) or not isinstance(port, int) or not 0 < port < 65536:
        raise ValueError("port is 1 to 65535, not %r" % (port,))
    if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not timeout > 0:
        raise ValueError("timeout must be positive, not %r" % (timeout,))
    if isinstance(retries, bool) or not isinstance(retries, int) or retries < 0:
        raise ValueError("retries is 0 or more, not %r" % (retries,))
    pktcap.replay_schedule(
        [], speed=speed, max_delay=max_delay, limit=limit
    )  # the arguments, before a file is read

    transfers = analyze(source, ports=ports, keep_payloads=writes).transfers
    skipped = 0
    steps: List[_Step] = []
    for transfer in sorted(transfers, key=lambda t: t.started):
        if transfer.operation == "write" and not (writes and _whole(transfer)):
            skipped += 1
        else:
            steps.append(_Step(transfer.started, transfer))
    results: List[TransferResult] = []
    for delay, step in pktcap.replay_schedule(steps, speed=speed, max_delay=max_delay, limit=limit):
        if delay:
            _sleep(delay)
        result = _run(step.transfer, host, port, float(timeout), retries)
        if result is None:
            skipped += 1
        else:
            results.append(result)
    return ReplayedTransfers(tuple(results), skipped)

summarize(data)

A one-line description of a TFTP datagram (never raises).

Text the peer chose (the mode, the options) is escaped, and a file name or an ERROR message is shown as repr shows it, so the line holds no control character.

Source code in src/tftp/capture/_events.py

def summarize(data: bytes) -> str:
    """A one-line description of a TFTP datagram (never raises).

    Text the peer chose (the mode, the options) is escaped, and a file name or an
    ERROR message is shown as ``repr`` shows it, so the line holds no control
    character.
    """
    if len(data) < 2:
        return "short datagram (%d bytes)" % len(data)
    op = (data[0] << 8) | data[1]
    if op == TFTPOpcode.DATA and len(data) >= 4:
        return "DATA %d (%d bytes)" % (struct.unpack_from("!H", data, 2)[0], len(data) - 4)
    if op == TFTPOpcode.ACK and len(data) >= 4:
        return "ACK %d" % struct.unpack_from("!H", data, 2)[0]
    try:
        packet = decode(data)
    except TFTPDecodeError as exc:
        return "malformed (%s)" % escape(str(exc))
    if isinstance(packet, RequestPacket):
        return "%s %r %s%s" % (
            TFTPOpcode(op).name,
            packet.filename,
            escape(packet.mode),
            _options(packet.options, " "),
        )
    if isinstance(packet, ErrorPacket):
        return "ERROR %d %r" % (packet.code, packet.message)
    if isinstance(packet, OptionAckPacket):
        return "OACK" + _options(packet.options, " ")
    return "opcode %d" % op

trace_to(writer)

A trace hook that appends every datagram it is given to writer.

writer stays the caller's to close. A write that raises (a full disk, a port or a time the format cannot hold) is the hook's failure: the transfer goes on and the first one is logged. ImportError when pktcap is not installed.

Source code in src/tftp/capture/_hook.py

def trace_to(writer: DatagramWriter) -> Callable[[PacketEvent], None]:
    """A trace hook that appends every datagram it is given to ``writer``.

    ``writer`` stays the caller's to close. A write that raises (a full disk, a
    port or a time the format cannot hold) is the hook's failure: the transfer
    goes on and the first one is logged. ``ImportError`` when pktcap is not installed.
    """
    require_pktcap()

    def hook(event: PacketEvent) -> None:
        writer.write(event.time, event.source, event.destination, event.data)

    return hook