`tftp.transfer`

The transfer engine: one side of a DATA/ACK exchange, without any I/O.

:class:Sender produces DATA and consumes ACKs; :class:Receiver consumes DATA and produces ACKs. :class:Requester is the client's opening: it sends the request, repeats it and judges the first answer. None touches a socket or a clock: packets go out through a send callable, packets come in through :meth:Transfer.handle (:meth:Requester.handle), and the driver calls on_timeout once deadline passes. The client and the server drive the same classes, and the tests drive them over a simulated lossy link.

Behaviour worth knowing:

Windowing (RFC 7440). A sender keeps a whole window of DATA in memory, so a retransmission never re-reads the source. An ACK inside the window restarts sending right after the acknowledged block.
Sorcerer's Apprentice (RFC 1123 4.2.3.1). A duplicate ACK never triggers a retransmission when windowsize is 1; only a timeout does. With a larger window the first duplicate of a given ACK resends the window (it is how a receiver reports a lost block) and later duplicates are ignored.
Block number rollover. Block numbers are tracked as unbounded integers and mapped to the 16-bit wire value, wrapping after 65535 to rollover (0 by default, 1 when negotiated), so file size is unlimited.

`Receiver(send, write, negotiated, retries, now, reply=None, complete=None, *, backoff=2.0, max_timeout=None, expires=None, max_idle=None)`

Bases: Transfer

Writes DATA packets to a sink and acknowledges them.

:param write: from :func:as_write. It may raise :class:WouldBlock: the block is then held unacknowledged until :meth:resume writes it, which is backpressure the sender sees as a slower ACK. :param reply: sent first and resent on timeout until DATA 1 arrives -- ACK 0 for a server answering a WRQ (or a client acknowledging an OACK), the OACK for a server answering a WRQ with options. None when the first DATA is about to be handed in (a client whose RRQ was answered with DATA 1 directly). :param complete: called after the last DATA is written and before it is acknowledged; if it raises, the peer gets an ERROR instead of the final ACK (a server committing an upload to disk). It may raise :class:WouldBlock too, holding the final ACK.

After the final ACK, is_done is set and the receiver keeps answering a repeated final DATA with that ACK, so a driver can "dally" (RFC 1350 section 6) by feeding it packets for a while longer.

Keyword arguments are :class:Transfer's (backoff, max_timeout, expires).

Source code in src/tftp/transfer/_receiver.py

def __init__(
    self,
    send: SendFunction,
    write: Callable[[Union[bytes, memoryview]], object],
    negotiated: Negotiated,
    retries: int,
    now: float,
    reply: Optional[bytes] = None,
    complete: Optional[Callable[[], object]] = None,
    *,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    expires: Optional[float] = None,
    max_idle: Optional[float] = None,
) -> None:
    super().__init__(
        send,
        negotiated,
        retries,
        backoff=backoff,
        max_timeout=max_timeout,
        expires=expires,
        max_idle=max_idle,
    )
    self._write = write
    self._complete = complete
    self._expected = 1
    self._expected_wire = 1
    self._in_window = 0
    self._nacks = 0  # ACKs sent for out-of-sequence DATA since the last in-order block
    self._reply = reply
    self._final: Optional[bytes] = None
    #: (payload, wire, written) of a block held by a stalled sink.
    self._pending: Optional[tuple] = None
    #: DATA arrived (and was dropped) while the sink was stalled.
    self._dropped = False
    #: A wire number above 1 arrived while block 65536 was missing.
    self._past_wire_one = False
    if reply is not None:
        send(reply)
        self._arm(now)

`Requester(send, server, opcode, filename, retries, now, *, mode='octet', options=None, timeout=1.0, backoff=2.0, max_timeout=None, expires=None, registry=None, accepts=None, fallback=True, probe=False)`

Sends a request, repeats it until the first answer arrives and judges that answer.

The request goes out from the constructor. The driver feeds it every datagram that arrives (:meth:handle) and calls :meth:on_timeout once deadline has passed; when is_done is set, error says whether the exchange failed and the outcome attributes say what the answer was. It reads no clock and touches no socket.

:param send: send(packet, address): the request goes to server, a refusal of the answer to the address it came from. May raise OSError: the outcome is recorded first, then the error propagates. :param server: where the request goes: (host, port, ...), opaque but for its first two items in a message. :param opcode: RRQ or WRQ. :param retries: repetitions of the request before giving up. :param now: the driver's clock when the request is sent. :param mode: "octet" or "netascii". :param options: the options the request carries, verbatim (None for none). :param timeout: seconds before the first repeat, and the timeout of a transfer that runs on the defaults. :param backoff: each consecutive repeat waits this many times longer. :param max_timeout: ceiling for the backed-off wait; None is eight times timeout. :param expires: the instant, in the driver's clock, after which the exchange fails whatever retries says, or None. :param registry: the :class:OptionRegistry an OACK is validated with. :param accepts: accepts(address): whether a datagram from address may be the answer. One that may not is dropped without a reply. None takes the first datagram from anywhere. :param fallback: allow :attr:retry_without_options. :param probe: the exchange only learns what the server answers: the answer is declined (ERROR 8 to an OACK, ACK 1 or ERROR 0 to DATA 1) and :attr:negotiated stays None.

:ivar is_done: the exchange has ended, in an answer or in a failure. :ivar error: the failure, or None: the :class:RemoteError of the server's code, :class:TFTPProtocolError for an answer no server may send or an OACK that cannot be accepted, :class:TransferTimeoutError. :ivar deadline: when :meth:on_timeout is due, in the driver's clock; None once the exchange has ended. :ivar peer: the address of the answer, or None before one. :ivar negotiated: what the answer settles (:class:Negotiated), or None on a failure and in a probe. :ivar first_data: the DATA 1 datagram when it was the answer to a read, else None. :ivar oack: the options of the OACK the server sent, as sent, or None. :ivar options: the options this request carried. :ivar retry_without_options: the answer was ERROR 8, 4 or 0 to a request that carried options and fallback is on: the request may be made again, by a new :class:Requester with options={}.

Source code in src/tftp/transfer/_requester.py

def __init__(
    self,
    send: Callable[[bytes, Address], object],
    server: Address,
    opcode: int,
    filename: str,
    retries: int,
    now: float,
    *,
    mode: str = "octet",
    options: Optional[Mapping[str, str]] = None,
    timeout: float = 1.0,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    expires: Optional[float] = None,
    registry: Optional[OptionRegistry] = None,
    accepts: Optional[Callable[[Address], bool]] = None,
    fallback: bool = True,
    probe: bool = False,
) -> None:
    from netimps import Backoff  # pure arithmetic: the engine still does no I/O

    self._send = send
    self._server = server
    self._is_read = opcode == TFTPOpcode.RRQ
    self._probe = probe
    self._retries = retries
    self._timeout = timeout
    self._registry = registry
    self._accepts = accepts
    self._fallback = fallback
    self.options: Dict[str, str] = dict(options or {})
    self._request = encode_request(opcode, filename, mode=mode, options=options)
    ceiling = max(timeout, max_timeout if max_timeout is not None else timeout * 8)
    self._timer = Backoff(timeout, multiplier=max(1.0, backoff), max_delay=ceiling)  # RFC 1123 4.2.3.2
    self._expires = expires
    self.is_done = False
    self.error: Optional[TFTPError] = None
    self.deadline: Optional[float] = None
    self.peer: Optional[Address] = None
    self.negotiated: Optional[Negotiated] = None
    self.first_data: Optional[bytes] = None
    self.oack: Optional[Dict[str, str]] = None
    self.retry_without_options = False
    self._arm(now)
    send(self._request, server)

`handle(packet, n, address, now)`

Take a datagram: it is the first answer, or it is dropped and nothing is sent.

Source code in src/tftp/transfer/_requester.py

def handle(self, packet: memoryview, n: int, address: Address, now: float) -> None:
    """Take a datagram: it is the first answer, or it is dropped and nothing is sent."""
    if self.is_done or n < 2 or (self._accepts is not None and not self._accepts(address)):
        return
    self.peer = address
    self.deadline = None
    self.is_done = True
    op = packet[1] if packet[0] == 0 else -1
    try:
        if op == TFTPOpcode.ERROR:
            self._refused(ErrorPacket.decode(packet[:n]))
            return
        if op == TFTPOpcode.OACK:
            self._oack(OptionAckPacket.decode(packet[:n]).options)
            return
    except TFTPDecodeError as exc:
        self.error = TFTPProtocolError("malformed answer to the request: %s" % exc)
        self.error.__cause__ = exc
        if op != TFTPOpcode.ERROR:
            self._reply(_encode_error(TFTPErrorCode.ILLEGAL_OPERATION, "malformed answer"))
        return
    if self._is_first_packet(packet, n):
        if self._probe:
            self.first_data = bytes(packet[:n])
            if n - 4 < DEFAULT_BLKSIZE:
                self._reply(encode_ack(1))  # the whole file: finish politely
            else:
                self._reply(_encode_error(TFTPErrorCode.NOT_DEFINED, "size probe only"))
            return
        self.negotiated = Negotiated(timeout=self._timeout)
        if self._is_read:
            self.first_data = bytes(packet[:n])
        return
    self.error = TFTPProtocolError("unexpected opcode %d in response to the request" % op)
    self._reply(_encode_error(TFTPErrorCode.ILLEGAL_OPERATION, "unexpected opcode %d" % op))

`on_timeout(now)`

Repeat the request, or fail when the time limit or the retries are spent.

Source code in src/tftp/transfer/_requester.py

def on_timeout(self, now: float) -> None:
    """Repeat the request, or fail when the time limit or the retries are spent."""
    if self.is_done:
        return
    if self._expires is not None and now >= self._expires:
        self._fail(TransferTimeoutError("transfer exceeded its time limit"))
        return
    timer = self._timer
    if timer.attempt >= self._retries:
        self._fail(TransferTimeoutError("no response from %s:%s" % self._server[:2]))
        return
    timer.advance()
    self._arm(now)
    self._send(self._request, self._server)

`Sender(send, read, negotiated, retries, now, oack=None, *, backoff=2.0, max_timeout=None, expires=None, max_idle=None)`

Bases: Transfer

Sends a stream as DATA packets: a server's RRQ, a client's WRQ.

:param read: a fill-the-view callable from :func:as_readinto. It may raise :class:WouldBlock; sending then pauses until :meth:resume. :param oack: when given, sent first, and DATA starts after ACK 0 (a server answering an RRQ that carried options). Otherwise DATA starts at once.

Keyword arguments are :class:Transfer's (backoff, max_timeout, expires).

Source code in src/tftp/transfer/_sender.py

def __init__(
    self,
    send: SendFunction,
    read: Callable[[memoryview], int],
    negotiated: Negotiated,
    retries: int,
    now: float,
    oack: Optional[bytes] = None,
    *,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    expires: Optional[float] = None,
    max_idle: Optional[float] = None,
) -> None:
    super().__init__(
        send,
        negotiated,
        retries,
        backoff=backoff,
        max_timeout=max_timeout,
        expires=expires,
        max_idle=max_idle,
    )
    self._read = read
    # Slots are allocated when their block is first read, so a transfer
    # costs what it has sent, not the window it negotiated.
    self._ring: list = [None] * self.windowsize
    self._views: list = [None] * self.windowsize
    self._lens = [0] * self.windowsize
    self._base = 1  # first block not yet acknowledged
    self._next = 1  # next block to transmit in the current window pass
    self._hi = 0  # highest block read into the ring
    self._last: Optional[int] = None  # the final block, once read
    self._resent_at = -(1 << 30)  # base when blocks in flight were last resent
    self._control = oack
    if oack is not None:
        send(oack)
        self._arm(now)
    else:
        self._pump(now)

`SupportsRead`

Bases: Protocol

A source with read(size) returning bytes (None: not ready).

`SupportsReadinto`

Bases: Protocol

A source with readinto(buffer), filling the buffer and returning the count (None: not ready).

`SupportsWrite`

Bases: Protocol

A sink with write(data): a binary file, or any object that takes bytes.

`Transfer(send, negotiated, retries, *, backoff=2.0, max_timeout=None, expires=None, max_idle=None)`

State shared by both directions.

:param retries: retransmissions of one packet before giving up. :param backoff: each consecutive retransmission waits this much longer (RFC 1123 4.2.3.2 asks for at least exponential backoff); progress resets the wait to the negotiated timeout. :param max_timeout: ceiling for the backed-off wait; None is eight times the timeout. :param expires: absolute time (driver's clock) by which the transfer must finish, or None. :param max_idle: seconds with no datagram from the peer after which the transfer fails, whatever timeout was negotiated; None is no bound. Time spent waiting on the local source or sink does not count.

:ivar is_done: the transfer is finished, successfully or not. :ivar error: the failure, or None. :ivar deadline: when :meth:on_timeout is due, in the driver's clock; None while nothing is outstanding. :ivar is_stalled: waiting for the source or sink (see :class:WouldBlock). :ivar bytes: payload bytes moved so far. :ivar retransmits: packets sent again after loss.