tftp.options

tftp.options

Option negotiation (RFC 2347): handlers, registry, server policy, profiles.

Every option is an :class:OptionHandler in an :class:OptionRegistry, so an application can add its own (vendor options, an option that selects content) without touching the transfer code. The RFC options (blksize, timeout, tsize, windowsize) are accepted by default; tftp-hpa's extensions (blksize2, utimeout, rollover, cookie) only when a server allows them, for example through a :class:Profile.

Blksize2Option

Bases: OptionHandler

blksize restricted to powers of two, for firmware that needs them.

ClientOptionContext(result, requested, is_read)

What a client-side option handler sees while checking an OACK.

Source code in src/tftp/options/_handler.py

def __init__(self, result: Negotiated, requested: Mapping[str, str], is_read: bool) -> None:
    self.result = result
    self.requested = requested
    self.is_read = is_read

CookieOption

Bases: OptionHandler

Opaque; echoed back unchanged. Its length is bounded by TFTPServerLimits.

MstfwindowOption

Bases: OptionHandler

Microsoft's variable-window extension (Windows 8+ bootmgr, WDS).

The client offers mstfwindow=31416; a server that speaks it answers 27182 and both start with a window of 4 blocks. The client may then resize the window through its ACKs, in a format Microsoft has not published; this library keeps the window at 4 (bytes after an ACK's block number are ignored). A windowsize negotiated alongside wins.

Negotiated(*, blksize=DEFAULT_BLKSIZE, windowsize=1, timeout=1.0, tsize=None, rollover=0, options=None, extra=None)

The parameters a transfer actually runs with, after negotiation.

options is what was acknowledged (the OACK contents), empty when the transfer runs on RFC 1350 defaults. extra holds what custom option handlers decided, keyed by option name.

Source code in src/tftp/options/_handler.py

def __init__(
    self,
    *,
    blksize: int = DEFAULT_BLKSIZE,
    windowsize: int = 1,
    timeout: float = 1.0,
    tsize: Optional[int] = None,
    rollover: int = 0,
    options: Optional[Dict[str, str]] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> None:
    self.blksize = blksize
    self.windowsize = windowsize
    self.timeout = timeout
    self.tsize = tsize
    self.rollover = rollover
    self.options = options if options is not None else {}
    self.extra = extra if extra is not None else {}

OptionHandler

One option's rules, for both ends of the negotiation.

Subclass and register an instance (see :class:OptionRegistry) to add an option. name is lower-case; standard marks the RFC-defined options a server accepts by default.

accept(requested, acked, ctx)

Client side: check the server's answer and apply it.

Raise :func:refuse for a value the RFC does not allow; the client then sends ERROR 8 and abandons the transfer.

Source code in src/tftp/options/_handler.py

def accept(self, requested: str, acked: str, ctx: ClientOptionContext) -> None:
    """Client side: check the server's answer and apply it.

    Raise :func:`refuse` for a value the RFC does not allow; the client
    then sends ERROR 8 and abandons the transfer.
    """
    ctx.result.extra[self.name] = acked

negotiate(value, ctx)

Server side: the value to acknowledge, or None to leave it out.

Leaving an option out is how RFC 2347 refuses it; the transfer then runs as if it had not been requested. Set fields on ctx.result.

Source code in src/tftp/options/_handler.py

def negotiate(self, value: str, ctx: ServerOptionContext) -> Optional[str]:
    """Server side: the value to acknowledge, or ``None`` to leave it out.

    Leaving an option out is how RFC 2347 refuses it; the transfer then
    runs as if it had not been requested. Set fields on ``ctx.result``.
    """
    return None

OptionRegistry(handlers=BUILTIN_OPTIONS)

Option handlers by name, kept in negotiation order.

Registration order is negotiation order, so an option whose answer depends on another (windowsize on the block size) is registered after it.

Source code in src/tftp/options/_registry.py

def __init__(self, handlers: Iterable[OptionHandler] = BUILTIN_OPTIONS) -> None:
    self._handlers: Dict[str, OptionHandler] = {}
    for handler in handlers:
        self.register(handler)

register(handler, *, replace=False)

Add handler; an existing name needs replace=True.

Source code in src/tftp/options/_registry.py

def register(self, handler: OptionHandler, *, replace: bool = False) -> OptionHandler:
    """Add ``handler``; an existing name needs ``replace=True``."""
    name = handler.name.lower()
    if not name:
        raise ValueError("an option handler needs a name")
    if handler.name != name:
        raise ValueError("option name %r must be lower-case, as a request's names are" % handler.name)
    if name in self._handlers and not replace:
        raise ValueError("option %r is already registered" % name)
    self._handlers[name] = handler
    return handler

standard()

Names of the RFC-defined options (what a server allows by default).

Source code in src/tftp/options/_registry.py

def standard(self) -> FrozenSet[str]:
    """Names of the RFC-defined options (what a server allows by default)."""
    return frozenset(h.name for h in self._handlers.values() if h.standard)

Profile(name, server, client)

A server policy and client settings that belong together.

Source code in src/tftp/options/_profiles.py

def __init__(self, name: str, server: TFTPServerOptions, client: Mapping[str, Any]) -> None:
    self.name = name
    self._server = copy.copy(server)
    self._client = dict(client)

client property

Keyword arguments for TFTPClient (a fresh dict each time).

server property

The server policy (a fresh copy each time, so changing it changes no profile).

ServerOptionContext(result, requested, policy, is_read, size=None, mtu=None, ipv6=False, stream=None)

What a server-side option handler sees while negotiating one request.

:ivar result: the :class:Negotiated being built; handlers set fields on it. :ivar requested: every requested option (lower-case names). :ivar acked: what has been acknowledged so far, in registry order. :ivar policy: the server's :class:TFTPServerOptions. :ivar is_read: RRQ (True) or WRQ. :ivar size: the file size for an RRQ (tsize), or None. :ivar mtu: link MTU of the interface the request arrived on, when known. :ivar ipv6: the transfer runs over IPv6 (header size for MTU arithmetic). :ivar stream: what the handler opened for an RRQ (options may describe it: x-mtime, x-list), else None.

Source code in src/tftp/options/_handler.py

def __init__(
    self,
    result: Negotiated,
    requested: Mapping[str, str],
    policy: Any,
    is_read: bool,
    size: Optional[int] = None,
    mtu: Optional[int] = None,
    ipv6: bool = False,
    stream: Any = None,
) -> None:
    self.result = result
    self.requested = requested
    self.acked = result.options
    self.policy = policy
    self.is_read = is_read
    self.size = size
    self.mtu = mtu
    self.ipv6 = ipv6
    self.stream = stream

max_blksize property

The policy's max_blksize, lowered to fit the MTU when asked to.

TFTPServerOptions(*, max_blksize=MAX_BLKSIZE, max_windowsize=64, max_window_bytes=4 << 20, allowed=None, refused=(), fit_mtu=False, registry=None)

A server's negotiation policy.

:param max_blksize: largest blksize granted; a larger request is answered with this value, which RFC 2348 allows. :param max_windowsize: largest windowsize granted. :param max_window_bytes: windowsize is also lowered so that one window of the negotiated block size fits in this many bytes -- the memory a sender holds per transfer. :param allowed: option names acknowledged at all. Defaults to the RFC options (blksize, timeout, tsize, windowsize); add :data:EXTENSION_OPTIONS names to accept extensions. :param refused: names never acknowledged even when allowed -- for firmware that asks for an option and then mishandles it. :param fit_mtu: lower blksize so a DATA packet fits the MTU of the interface the request arrived on (needs pktinfo). Boot ROMs often cannot reassemble IP fragments. :param registry: where option handlers come from; custom options are registered there.

Source code in src/tftp/options/_policy.py

def __init__(
    self,
    *,
    max_blksize: int = MAX_BLKSIZE,
    max_windowsize: int = 64,
    max_window_bytes: int = 4 << 20,
    allowed: Optional[Iterable[str]] = None,
    refused: Iterable[str] = (),
    fit_mtu: bool = False,
    registry: Optional[OptionRegistry] = None,
) -> None:
    self.registry = registry or DEFAULT_REGISTRY
    if not MIN_BLKSIZE <= max_blksize <= MAX_BLKSIZE:
        raise ValueError("max_blksize must be in %d..%d" % (MIN_BLKSIZE, MAX_BLKSIZE))
    if not 1 <= max_windowsize <= MAX_WINDOWSIZE:
        raise ValueError("max_windowsize must be in 1..%d" % MAX_WINDOWSIZE)
    if max_window_bytes < MIN_BLKSIZE:
        raise ValueError("max_window_bytes is smaller than any block")
    allowed_set = (
        frozenset(n.lower() for n in allowed) if allowed is not None else self.registry.standard()
    )
    unknown = allowed_set - self.registry.names()
    if unknown:
        raise ValueError("unsupported options: %s" % ", ".join(sorted(unknown)))
    self.max_blksize = max_blksize
    self.max_windowsize = max_windowsize
    self.max_window_bytes = max_window_bytes
    self.allowed = allowed_set
    self.refused = frozenset(n.lower() for n in refused)
    self.fit_mtu = fit_mtu

replace(**changes)

A new policy with every field of this one, except those in changes.

As :func:dataclasses.replace: the result is built through the constructor, so it is validated, and this policy is not modified.

:raises TypeError: a name that is not a field.

Source code in src/tftp/options/_policy.py

def replace(self, **changes: Any) -> "TFTPServerOptions":
    """A new policy with every field of this one, except those in ``changes``.

    As :func:`dataclasses.replace`: the result is built through the
    constructor, so it is validated, and this policy is not modified.

    :raises TypeError: a name that is not a field.
    """
    unknown = sorted(set(changes) - set(self.__slots__))
    if unknown:
        raise TypeError("TFTPServerOptions has no field %s" % ", ".join(repr(n) for n in unknown))
    fields = {name: getattr(self, name) for name in self.__slots__}
    fields.update(changes)
    return type(self)(**fields)

XListOption

Bases: OptionHandler

pytftp's directory listing (see :mod:tftp.listing): x-list=1.

Acknowledged only when the handler answered the RRQ with a listing (a stream with lists_directories = True); a client seeing it left out of the OACK knows it is receiving a file.

XMtimeOption

Bases: OptionHandler

pytftp's x-mtime: an RRQ's OACK carries the file's modification time (whole seconds since the epoch), when the source knows it.

accept_oack(requested, oack, *, is_read, timeout, registry=None)

Client side: validate a server's OACK against what was requested.

Raises :class:TFTPProtocolError with code 8 when the server acknowledged an option that was not requested, or answered outside what the option's RFC allows (a larger blksize or windowsize than asked for, ...). The caller sends that ERROR to the server and abandons the transfer. An option the server left out runs on its default.

Source code in src/tftp/options/_negotiate.py

def accept_oack(
    requested: Mapping[str, str],
    oack: Mapping[str, str],
    *,
    is_read: bool,
    timeout: float,
    registry: Optional[OptionRegistry] = None,
) -> Negotiated:
    """Client side: validate a server's OACK against what was requested.

    Raises :class:`TFTPProtocolError` with code 8 when the server acknowledged an
    option that was not requested, or answered outside what the option's RFC
    allows (a larger ``blksize`` or ``windowsize`` than asked for, ...). The
    caller sends that ERROR to the server and abandons the transfer. An
    option the server left out runs on its default.
    """
    registry = registry or DEFAULT_REGISTRY
    result = Negotiated(timeout=timeout, options=dict(oack))
    ctx = ClientOptionContext(result, requested, is_read)
    for name, value in oack.items():
        if name not in requested:
            raise refuse("server acknowledged unrequested option %r" % name)
        handler = registry.get(name)
        if handler is None:
            result.extra[name] = value  # a custom option the caller asked for
        else:
            handler.accept(requested[name], value, ctx)
    return result

negotiate(requested, policy, *, is_read, timeout, size=None, mtu=None, ipv6=False, stream=None)

Server side: decide which requested options to acknowledge.

Options the server does not know, does not allow, or that carry an unusable value are left out of the acknowledgement -- RFC 2347's way of refusing them; the transfer then runs on that option's default. A server never acknowledges an option that was not requested. size is the file size for an RRQ tsize; mtu/ipv6 feed fit_mtu; stream is the RRQ's opened source, for options that describe it.

Source code in src/tftp/options/_negotiate.py

def negotiate(
    requested: Mapping[str, str],
    policy: TFTPServerOptions,
    *,
    is_read: bool,
    timeout: float,
    size: Optional[int] = None,
    mtu: Optional[int] = None,
    ipv6: bool = False,
    stream: Any = None,
) -> Negotiated:
    """Server side: decide which requested options to acknowledge.

    Options the server does not know, does not allow, or that carry an
    unusable value are left out of the acknowledgement -- RFC 2347's way of
    refusing them; the transfer then runs on that option's default. A server
    never acknowledges an option that was not requested. ``size`` is the file
    size for an RRQ ``tsize``; ``mtu``/``ipv6`` feed ``fit_mtu``; ``stream``
    is the RRQ's opened source, for options that describe it.
    """
    result = Negotiated(timeout=timeout)
    ctx = ServerOptionContext(result, requested, policy, is_read, size, mtu, ipv6, stream)
    for handler in policy.registry:
        name = handler.name
        if name in requested and policy.accepts(name):
            answer = handler.negotiate(requested[name], ctx)
            if answer is not None:
                result.options[name] = answer
    return result

refuse(message)

The ERROR 8 a client sends for an OACK it cannot accept.

Source code in src/tftp/options/_handler.py

def refuse(message: str) -> TFTPProtocolError:
    """The ERROR 8 a client sends for an OACK it cannot accept."""
    return TFTPProtocolError(message, TFTPErrorCode.OPTION_REFUSED)

register_option(handler, replace=False)

Add handler to :data:DEFAULT_REGISTRY (usable as a class decorator's result).

Source code in src/tftp/options/_registry.py

def register_option(handler: OptionHandler, replace: bool = False) -> OptionHandler:
    """Add ``handler`` to :data:`DEFAULT_REGISTRY` (usable as a class decorator's result)."""
    return DEFAULT_REGISTRY.register(handler, replace=replace)

request_options(*, blksize=None, windowsize=None, timeout=None, tsize=None, rollover=None, utimeout=False, extra=None)

Client side: the options to put in a request.

timeout is sent as timeout when it is a whole number of seconds in 1..255; a fractional one is sent as utimeout only with utimeout=True (an extension) and otherwise not at all. tsize is 0 for a read and the upload size for a write; None leaves it out. extra adds options verbatim (custom or extension options), after the built-in ones.

Source code in src/tftp/options/_negotiate.py

def request_options(
    *,
    blksize: Optional[int] = None,
    windowsize: Optional[int] = None,
    timeout: Optional[float] = None,
    tsize: Optional[int] = None,
    rollover: Optional[int] = None,
    utimeout: bool = False,
    extra: Optional[Mapping[str, object]] = None,
) -> Dict[str, str]:
    """Client side: the options to put in a request.

    ``timeout`` is sent as ``timeout`` when it is a whole number of seconds in
    1..255; a fractional one is sent as ``utimeout`` only with
    ``utimeout=True`` (an extension) and otherwise not at all. ``tsize`` is
    ``0`` for a read and the upload size for a write; ``None`` leaves it out.
    ``extra`` adds options verbatim (custom or extension options), after the
    built-in ones.
    """
    options: Dict[str, str] = {}
    if blksize is not None:
        if not MIN_BLKSIZE <= blksize <= MAX_BLKSIZE:
            raise ValueError("blksize must be in %d..%d" % (MIN_BLKSIZE, MAX_BLKSIZE))
        options["blksize"] = str(blksize)
    if tsize is not None:
        if tsize < 0:
            raise ValueError("tsize cannot be negative")
        options["tsize"] = str(tsize)
    if timeout is not None:
        if timeout <= 0:
            raise ValueError("timeout must be positive")
        if float(timeout).is_integer() and 1 <= timeout <= 255:
            options["timeout"] = str(int(timeout))
        elif utimeout:
            micro = int(round(timeout * 1e6))
            options["utimeout"] = str(min(max(micro, MIN_UTIMEOUT), MAX_UTIMEOUT))
    if windowsize is not None:
        if not 1 <= windowsize <= MAX_WINDOWSIZE:
            raise ValueError("windowsize must be in 1..%d" % MAX_WINDOWSIZE)
        options["windowsize"] = str(windowsize)
    if rollover is not None:
        if rollover not in (0, 1):
            raise ValueError("rollover must be 0 or 1")
        options["rollover"] = str(rollover)
    for name, value in (extra or {}).items():
        key = str(name).lower()
        if key in options:
            raise ValueError("option %r given twice" % key)
        options[key] = str(value)
    return options