tftp.packet

tftp.packet

The TFTP wire format: opcodes, error codes, packet types, encode/decode.

AckPacket(block) dataclass

Bases: _Packet

ACK of block block (0 to 65535).

decode(data) classmethod

The ACK data holds; :class:TFTPDecodeError for another packet.

Source code in src/tftp/packet/_codec.py

@classmethod
def decode(cls, data: Union[bytes, bytearray, memoryview]) -> "AckPacket":
    """The ACK ``data`` holds; :class:`TFTPDecodeError` for another packet."""
    return _decode_kind(cls, data)

DataPacket(block, data) dataclass

Bases: _Packet

DATA: block number block (0 to 65535) carrying data.

decode(data) classmethod

The DATA data holds; :class:TFTPDecodeError for another packet.

Source code in src/tftp/packet/_codec.py

@classmethod
def decode(cls, data: Union[bytes, bytearray, memoryview]) -> "DataPacket":
    """The DATA ``data`` holds; :class:`TFTPDecodeError` for another packet."""
    return _decode_kind(cls, data)

ErrorPacket(code, message) dataclass

Bases: _Packet

ERROR: code is a :class:TFTPErrorCode, which carries a number this library does not define (any 16-bit value) as an unnamed member.

decode(data) classmethod

The ERROR data holds; :class:TFTPDecodeError for another packet.

Source code in src/tftp/packet/_codec.py

@classmethod
def decode(cls, data: Union[bytes, bytearray, memoryview]) -> "ErrorPacket":
    """The ERROR ``data`` holds; :class:`TFTPDecodeError` for another packet."""
    return _decode_kind(cls, data)

encode()

The strict encoding, :func:encode_error.

Source code in src/tftp/packet/_codec.py

def encode(self) -> bytes:
    """The strict encoding, :func:`encode_error`."""
    return encode_error(self.code, self.message)

OptionAckPacket(options) dataclass

Bases: _Packet

OACK: the options a server accepted, a read-only mapping of lower-cased names to text values.

decode(data) classmethod

The OACK data holds; :class:TFTPDecodeError for another packet.

Source code in src/tftp/packet/_codec.py

@classmethod
def decode(cls, data: Union[bytes, bytearray, memoryview]) -> "OptionAckPacket":
    """The OACK ``data`` holds; :class:`TFTPDecodeError` for another packet."""
    return _decode_kind(cls, data)

RequestPacket(opcode, filename, mode, options=dict(), raw=b'') dataclass

Bases: _Packet

A read (RRQ) or write (WRQ) request.

options is a read-only mapping with lower-cased names, as RFC 2347 makes option names case-insensitive; values are text, kept as sent, in request order. raw is the datagram exactly as received (empty for a request built in code), so a relay can forward it untouched -- option spelling, order, duplicates and options this library does not know all survive. raw is not part of equality. mode is lower-cased and not checked, because a request is decoded as the peer sent it; :meth:encode refuses a mode RFC 1350 does not define.

raw_options property

Every option pair as sent: original case, order and duplicates.

decode(data) classmethod

The request data holds; :class:TFTPDecodeError for another packet.

Source code in src/tftp/packet/_codec.py

@classmethod
def decode(cls, data: Union[bytes, bytearray, memoryview]) -> "RequestPacket":
    """The request ``data`` holds; :class:`TFTPDecodeError` for another packet."""
    return _decode_kind(cls, data)

encode()

raw when there is one, else a fresh encoding (:func:encode_request).

Source code in src/tftp/packet/_codec.py

def encode(self) -> bytes:
    """``raw`` when there is one, else a fresh encoding (:func:`encode_request`)."""
    return self.raw or encode_request(self.opcode, self.filename, mode=self.mode, options=self.options)

TFTPErrorCode

Bases: IntEnum

ERROR packet codes (RFC 1350 section 5, plus RFC 2347's code 8).

TFTPErrorCode(n) for any n in 0..65535 the RFCs do not define is an unnamed member (name is CODE_<n>) that carries the number, so a peer's code is forwarded as it came; ValueError outside that range.

TFTPOpcode

Bases: IntEnum

Packet type, the first two bytes of every packet.

decode(packet)

Parse one packet, whichever kind it is.

Raises :class:TFTPDecodeError on invalid input. Liberal where real peers are: a missing final NUL, a dangling option name, an unknown mode and an unknown error code are accepted.

Source code in src/tftp/packet/_codec.py

def decode(packet: Union[bytes, bytearray, memoryview]) -> TFTPPacket:
    """Parse one packet, whichever kind it is.

    Raises :class:`TFTPDecodeError` on invalid input. Liberal where real peers
    are: a missing final NUL, a dangling option name, an unknown mode and an
    unknown error code are accepted.
    """
    buf = bytes(packet)
    if len(buf) < 2:
        raise TFTPDecodeError("packet shorter than an opcode")
    (op,) = _OP.unpack_from(buf)
    if op in (TFTPOpcode.RRQ, TFTPOpcode.WRQ):
        fields = _strings(buf[2:])
        if len(fields) < 2 or not fields[0]:
            raise TFTPDecodeError("request without a filename and mode")
        return RequestPacket(TFTPOpcode(op), fields[0], fields[1], _parse_options(fields[2:]), buf)
    if op == TFTPOpcode.DATA:
        if len(buf) < 4:
            raise TFTPDecodeError("DATA shorter than its header")
        return DataPacket(_HDR.unpack_from(buf)[1], buf[4:])
    if op == TFTPOpcode.ACK:
        if len(buf) < 4:
            raise TFTPDecodeError("ACK shorter than its header")
        return AckPacket(_HDR.unpack_from(buf)[1])
    if op == TFTPOpcode.ERROR:
        if len(buf) < 4:
            raise TFTPDecodeError("ERROR shorter than its header")
        return ErrorPacket(_HDR.unpack_from(buf)[1], _error_text(buf[4:]))
    if op == TFTPOpcode.OACK:
        return OptionAckPacket(_parse_options(_strings(buf[2:])))
    raise TFTPDecodeError("unknown opcode %d" % op)

encode_ack(block)

Build an ACK. ValueError for a block outside 0..65535, TypeError for a non-int.

Source code in src/tftp/packet/_codec.py

def encode_ack(block: int) -> bytes:
    """Build an ACK. ``ValueError`` for a block outside 0..65535, ``TypeError`` for a non-``int``."""
    _check_block(block)
    return _HDR.pack(TFTPOpcode.ACK, block)

encode_data(block, data)

Build a DATA. ValueError for a block outside 0..65535, TypeError for a non-int.

Source code in src/tftp/packet/_codec.py

def encode_data(block: int, data: bytes) -> bytes:
    """Build a DATA. ``ValueError`` for a block outside 0..65535, ``TypeError`` for a non-``int``."""
    _check_block(block)
    return _HDR.pack(TFTPOpcode.DATA, block) + bytes(data)

encode_error(code, message='')

Build an ERROR.

Raises ValueError for a code outside 0..65535, a NUL in message and a message over 512 octets once encoded; TypeError for a code that is not an int or a message that is not text.

Source code in src/tftp/packet/_codec.py

def encode_error(code: int, message: str = "") -> bytes:
    """Build an ERROR.

    Raises ``ValueError`` for a ``code`` outside 0..65535, a NUL in
    ``message`` and a message over 512 octets once encoded; ``TypeError`` for
    a ``code`` that is not an ``int`` or a ``message`` that is not text.
    """
    if isinstance(code, bool) or not isinstance(code, int):
        raise TypeError("an error code is an int, not %s" % type(code).__name__)
    if not 0 <= code <= 65535:
        raise ValueError("an error code is 0 to 65535, not %d" % code)
    raw = _text(message, "an error message")
    if len(raw) > MAX_ERROR_TEXT:
        raise ValueError("an error message is at most %d octets, this one is %d" % (MAX_ERROR_TEXT, len(raw)))
    return _HDR.pack(TFTPOpcode.ERROR, code) + raw + b"\0"

encode_oack(options)

Build an OACK. ValueError for no options, an empty option name or a NUL; TypeError for a value that is neither text nor an int.

Source code in src/tftp/packet/_codec.py

def encode_oack(options: Mapping[str, object]) -> bytes:
    """Build an OACK. ``ValueError`` for no options, an empty option name or a
    NUL; ``TypeError`` for a value that is neither text nor an ``int``."""
    if not options:
        raise ValueError("an OACK carries at least one option")
    return _OP.pack(TFTPOpcode.OACK) + _options_bytes(options)

encode_request(opcode, filename, *, mode='octet', options=None)

Build an RRQ or WRQ.

Option values may be text or an int, sent in decimal. Raises ValueError for an opcode that is not RRQ or WRQ, an empty file name or option name, a NUL in any string, a mode other than octet, netascii or mail (any case) and a request over the 512 octets RFC 2347 allows; TypeError for an argument of the wrong type.

Source code in src/tftp/packet/_codec.py

def encode_request(
    opcode: int,
    filename: str,
    *,
    mode: str = "octet",
    options: Optional[Mapping[str, object]] = None,
) -> bytes:
    """Build an RRQ or WRQ.

    Option values may be text or an ``int``, sent in decimal. Raises
    ``ValueError`` for an opcode that is not RRQ or WRQ, an empty file name or
    option name, a NUL in any string, a ``mode`` other than ``octet``,
    ``netascii`` or ``mail`` (any case) and a request over the 512 octets
    RFC 2347 allows; ``TypeError`` for an argument of the wrong type.
    """
    if opcode not in (TFTPOpcode.RRQ, TFTPOpcode.WRQ):
        raise ValueError("a request opcode is RRQ or WRQ, not %r" % (opcode,))
    name = _text(filename, "a file name")
    if not name:
        raise ValueError("a request needs a file name")
    mode_text = _text(mode, "a mode")
    if mode.lower() not in _MODES:
        raise ValueError("a request mode is netascii, octet or mail, not %r" % (mode,))
    packet = _OP.pack(opcode) + name + b"\0" + mode_text + b"\0" + _options_bytes(options)
    if len(packet) > MAX_REQUEST_SIZE:
        raise ValueError("a request is at most %d octets, this one is %d" % (MAX_REQUEST_SIZE, len(packet)))
    return packet