tftp.server

tftp.server

The TFTP servers (blocking and asyncio), their request handlers, and the per-transfer session.

AsyncTFTPHandler

Bases: Protocol

What an :class:~tftp.AsyncTFTPServer needs from a handler: coroutine hooks.

async open_read(context) returns an :class:AsyncTFTPReader and async open_write(context, size) an :class:AsyncTFTPWriter. Both run as tasks on the server's event loop and may raise what :class:TFTPHandler's hooks may. A synchronous handler is given to the server through :class:ThreadedHandler.

AsyncTFTPReader

Bases: Protocol

An asynchronous source an :class:~tftp.AsyncTFTPServer reads a download from.

Required: async read(size) returning b"" at the end, and async close(). Optional: size, the total length, which answers tsize. The server reads ahead into a bounded buffer, so a slow source slows the transfer instead of filling memory.

AsyncTFTPServer(root_or_handler, *, listen=None, port=69, writable=False, create=True, overwrite=False, timeout=1.0, retries=5, options=None, max_sessions=DEFAULT_MAX_SESSIONS, reply_from_request_address=True, dally=True, on_complete=None, limits=None, ignore_broadcast=True, backoff=2.0, max_timeout=None, trace=None, port_range=None)

Bases: AsyncioService, ServerBase

:class:tftp.TFTPServer for asyncio; same arguments except open_in_thread/workers.

root_or_handler is a directory, a handler with coroutine hooks (:class:AsyncTFTPHandler) or a synchronous handler wrapped in :class:ThreadedHandler; a bare synchronous handler raises TypeError. Use as async with AsyncTFTPServer(...) as server: await server.serve_forever(), or await server.start() ... await server.aclose().

The constructor opens nothing: :meth:bind does, and :meth:start, :meth:serve_forever and async with call it. shutdown() never blocks and is safe from any thread; await wait_closed() returns once serving has stopped; await aclose() shuts down, waits and releases the sockets, and is final: :meth:start or :meth:serve_forever afterwards, or while serving, raises RuntimeError.

Source code in src/tftp/server/_asyncio.py

def __init__(
    self,
    root_or_handler: Union[str, "os.PathLike[str]", AsyncTFTPHandler, ThreadedHandler],
    *,
    listen: ListenLike = None,
    port: int = 69,
    writable: bool = False,
    create: bool = True,
    overwrite: bool = False,
    timeout: float = 1.0,
    retries: int = 5,
    options: Optional[TFTPServerOptions] = None,
    max_sessions: Optional[int] = DEFAULT_MAX_SESSIONS,
    reply_from_request_address: bool = True,
    dally: bool = True,
    on_complete: Optional[Callable[[TransferResult], Any]] = None,
    limits: Optional[TFTPServerLimits] = None,
    ignore_broadcast: bool = True,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    trace: Optional[Callable[[PacketEvent], Any]] = None,
    port_range: Optional[PortRangeLike] = None,
) -> None:
    is_root = isinstance(root_or_handler, (str, os.PathLike))
    if (
        not is_root
        and not isinstance(root_or_handler, ThreadedHandler)
        and not has_coroutine_hooks(root_or_handler)
    ):
        raise TypeError(
            "%s has plain hooks; AsyncTFTPServer takes coroutine hooks, or a synchronous handler "
            "wrapped as ThreadedHandler(handler)" % type(root_or_handler).__name__
        )
    super().__init__(
        root_or_handler,
        listen=listen,
        port=port,
        writable=writable,
        create=create,
        overwrite=overwrite,
        timeout=timeout,
        retries=retries,
        options=options,
        max_sessions=max_sessions,
        reply_from_request_address=reply_from_request_address,
        dally=dally,
        on_complete=on_complete,
        limits=limits,
        ignore_broadcast=ignore_broadcast,
        backoff=backoff,
        max_timeout=max_timeout,
        trace=trace,
        port_range=port_range,
    )
    if is_root:
        self.handler = ThreadedHandler(self.handler)
    # Only the adapter returns synchronous streams; a coroutine handler returns asynchronous ones.
    self._async_streams = not isinstance(self.handler, ThreadedHandler)
    self._init_service()

AsyncTFTPWriter

Bases: Protocol

An asynchronous sink an :class:~tftp.AsyncTFTPServer writes an upload into.

Required: async write(data) and async close(), which is awaited before the last block is acknowledged, so a failing close reaches the client as an ERROR. The server buffers a bounded amount ahead of the sink; data is always bytes. An abandoned upload cancels the write in progress and does not call close.

AtomicWriter(path, *, overwrite=True, mode=None)

Writes to a temporary file beside path and renames it on close.

A partial upload therefore never replaces or appears as path; :meth:abort deletes the temporary file. The temporary file is private (mode 0600) unless mode is given: then it is created with that mode less the process's umask, and takes the permissions of the file it replaces. A process killed mid-transfer leaves .name.*.part beside path. On Windows a refused rename is tried again for half a second.

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

def __init__(self, path: str, *, overwrite: bool = True, mode: Optional[int] = None) -> None:
    self.path = path
    self.overwrite = overwrite
    directory, name = os.path.split(path)
    if mode is None:
        fd, self._tmp = tempfile.mkstemp(prefix=".%s." % name, suffix=".part", dir=directory or ".")
    else:
        fd, self._tmp = _create_beside(directory or ".", name, mode)
        try:
            os.chmod(self._tmp, stat.S_IMODE(os.stat(path).st_mode))
        except OSError:  # nothing to take the permissions of
            pass
    self._file = os.fdopen(fd, "wb")
    self.closed = False

PortRange(low, high) dataclass

The UDP ports transfer sockets may use, low to high inclusive.

Each transfer needs a port of its own; pinning them to a range lets a firewall allow them (tftp-hpa -R, dnsmasq --tftp-port-range). An immutable value: len() counts the ports, iteration and in cover them in order, str() is LOW:HIGH and PortRange.parse reads it back. Which port a server takes next is the server's own :class:PortAllocator.

:raises TypeError: a bound that is not an int. :raises TFTPValueError: not 1 <= low <= high <= 65535.

parse(text) classmethod

LOW:HIGH (LOW-HIGH too), ASCII digits, as str(range) writes it.

:raises TypeError: text is not a str. :raises TFTPValueError: anything else.

Source code in src/tftp/server/_session.py

@classmethod
def parse(cls, text: str) -> "PortRange":
    """``LOW:HIGH`` (``LOW-HIGH`` too), ASCII digits, as ``str(range)`` writes it.

    :raises TypeError: ``text`` is not a ``str``.
    :raises TFTPValueError: anything else.
    """
    if not isinstance(text, str):
        raise TypeError("a port range is parsed from text, not %s" % type(text).__name__)
    low, sep, high = text.replace("-", ":", 1).partition(":")
    if not sep or not all(part.isascii() and part.isdigit() for part in (low, high)):
        raise TFTPValueError("a port range is LOW:HIGH within 1..65535, not %r" % text)
    return cls(int(low), int(high))

try_parse(text, default=None) classmethod

:meth:parse, or default for text that is not a port range.

Still raises :class:TypeError when text is not a str.

Source code in src/tftp/server/_session.py

@classmethod
def try_parse(cls, text: str, default: "Optional[PortRange]" = None) -> "Optional[PortRange]":
    """:meth:`parse`, or ``default`` for text that is not a port range.

    Still raises :class:`TypeError` when ``text`` is not a ``str``.
    """
    try:
        return cls.parse(text)
    except TFTPValueError:
        return default

TFTPChunkReader

Bases: Protocol

A :class:TFTPReader that has read(size) in place of readinto.

TFTPHandler

Bases: Protocol

What a synchronous server needs from a handler.

Both hooks run one request at a time, in a worker thread unless the handler has opens_fast = True (an optional attribute: opening is quick and never blocks, so it runs inline on the server's thread). open_read returns a :class:TFTPReader or :class:TFTPChunkReader; open_write returns a :class:TFTPWriter, given the client's announced tsize as size (or None).

Either hook may raise :class:TFTPError to refuse with a specific code, or OSError, which is mapped by errno (ENOENT -> file not found, EACCES -> access violation, ENOSPC -> disk full...).

A hook is a plain function: a handler whose hooks are async def belongs to :class:~tftp.AsyncTFTPServer, and the synchronous server refuses it.

TFTPReader

Bases: Protocol

A binary source a synchronous server reads a download from.

Required: readinto(buffer) and close(). A source with read(size) instead is a :class:TFTPChunkReader.

Optional members, each read once per transfer when present:

• size -- the total length, an integer, which answers tsize;
• mtime -- the modification time in seconds, which answers x-mtime;
• set_wakeup(callback) -- stores a callable the source calls, from any thread, when a readinto that raised :class:tftp.WouldBlock can make progress;
• lists_directories -- True on a source that holds a directory listing, which is what makes the server acknowledge x-list.

readinto may raise :class:tftp.WouldBlock for "no data yet".

TFTPRequestContext(request, peer, *, local_address=None, interface_index=0)

Everything known about a request when a handler is asked to open it.

:ivar request: the parsed RRQ/WRQ (filename, mode, options). :ivar peer: the client's (host, port, ...). :ivar local_address: the address the request was sent to, when the platform reports it (see TFTPServer.has_pktinfo), else None. :ivar interface_index: the interface it arrived on, or 0. :ivar interface: that interface as a netimps.Interface (name, addresses, MTU), or None when unknown. :ivar listing: an RRQ asking for a directory listing (x-list) that the server's policy allows; a handler may then answer a directory with a :class:tftp.listing.DirectoryListing.

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

def __init__(
    self,
    request: RequestPacket,
    peer: Tuple[Any, ...],
    *,
    local_address: Optional[str] = None,
    interface_index: int = 0,
) -> None:
    self.request = request
    self.peer = peer
    self.local_address = local_address
    self.interface_index = interface_index
    self.interface: Optional[Interface] = None
    self.listing = False

with_filename(filename)

A copy of this context for a request naming filename instead.

Everything else is carried over (the listing flag, the interface), so a wrapper that rewrites names does not have to know the fields.

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

def with_filename(self, filename: str) -> "TFTPRequestContext":
    """A copy of this context for a request naming ``filename`` instead.

    Everything else is carried over (the listing flag, the interface), so
    a wrapper that rewrites names does not have to know the fields.
    """
    clone = copy.copy(self)
    clone.request = dataclasses.replace(self.request, filename=filename, raw=b"")
    return clone

TFTPServer(root_or_handler, *, listen=None, port=69, writable=False, create=True, overwrite=False, timeout=1.0, retries=5, options=None, max_sessions=DEFAULT_MAX_SESSIONS, reply_from_request_address=True, dally=True, on_complete=None, limits=None, ignore_broadcast=True, backoff=2.0, max_timeout=None, trace=None, open_in_thread=None, workers=8, port_range=None)

Bases: SelectorService, ServerBase

A TFTP server.

:param root_or_handler: a directory to serve (wrapped in :class:FilesystemBackend with writable, create and overwrite), or a handler object. :param listen: where to listen, as netimps reads it (netimps.ListenLike): an address, "host:port", "[v6]:port", an adapter name or MAC (a wildcard socket limited to that adapter), several of them in a list or joined by commas. None (the default) is every address of both families: "*" and "::", two sockets, the IPv6 one left out on a host with no IPv6. A host name is never resolved. :param port: the port a listen entry that names none gets; 0 picks a free one (with listen=None both sockets share it; see :attr:server_address and :attr:server_addresses, valid after :meth:bind). :param timeout: retransmission timeout, unless a client negotiates its own. :param retries: retransmissions of one packet before abandoning a transfer. :param options: what the server negotiates (:class:TFTPServerOptions). :param max_sessions: concurrent transfers; requests beyond it get ERROR 0 "server busy". The default is 500 on every platform; None is unlimited, except on Windows, where select() caps it at 510 (a larger number there is a ValueError). :param reply_from_request_address: answer each request from the address it was sent to (pktinfo). When the platform cannot report it, replies come from the listening address, or the routing table's choice when listening on a wildcard. :param dally: after acknowledging the last block of an upload, keep the transfer open for one timeout to re-acknowledge a retransmitted last block (RFC 1350 section 6). Costs nothing but a socket for that long. :param on_complete: called with a :class:TransferResult after every transfer, failed ones included (result.error is set). :param limits: request, per-client and duration bounds (:class:TFTPServerLimits). :param ignore_broadcast: silently drop requests sent to a broadcast or multicast address (RFC 1123 4.2.3.5). Needs pktinfo to see the destination; without it every request looks unicast. :param backoff: each consecutive retransmission waits this many times longer (RFC 1123 4.2.3.2); progress resets it. :param max_timeout: ceiling for the backed-off wait; None is eight times the timeout. :param trace: trace(PacketEvent) for every datagram of every transfer, received and sent (role="server", one session id per transfer). Requests refused before a transfer exists are not traced. :func:tftp.capture.trace_to makes a hook of a pktcap writer. :param open_in_thread: call the handler's open_read/open_write in a worker thread, so a handler that blocks (an HTTP request, an upstream server) never stalls other transfers. None decides from the handler: those with opens_fast = True (the built-in file and memory handlers) open inline, everything else in a worker. :param workers: size of that worker pool. :param port_range: a :class:PortRange, a (low, high) pair (inclusive), a range or "LOW:HIGH" text: transfer sockets take their ports from it, round-robin from a position this server owns, so a firewall can allow them. A request arriving while every port is in use gets ERROR 0 "server busy". None lets the OS choose.

Source code in src/tftp/server/_sync.py

def __init__(
    self,
    root_or_handler: Union[str, "os.PathLike[str]", TFTPHandler],
    *,
    listen: ListenLike = None,
    port: int = 69,
    writable: bool = False,
    create: bool = True,
    overwrite: bool = False,
    timeout: float = 1.0,
    retries: int = 5,
    options: Optional[TFTPServerOptions] = None,
    max_sessions: Optional[int] = DEFAULT_MAX_SESSIONS,
    reply_from_request_address: bool = True,
    dally: bool = True,
    on_complete: Optional[Callable[[TransferResult], Any]] = None,
    limits: Optional[TFTPServerLimits] = None,
    ignore_broadcast: bool = True,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    trace: Optional[Callable[[PacketEvent], Any]] = None,
    open_in_thread: Optional[bool] = None,
    workers: int = 8,
    port_range: Optional[PortRangeLike] = None,
) -> None:
    if has_coroutine_hooks(root_or_handler):
        raise TypeError(
            "%r has coroutine hooks; TFTPServer takes plain functions (AsyncTFTPServer takes coroutines)"
            % (root_or_handler,)
        )
    if sys.platform == "win32" and max_sessions is not None and max_sessions > SELECT_SESSIONS:
        raise ValueError(
            "max_sessions=%d is more than the %d transfers select() can watch on Windows"
            % (max_sessions, SELECT_SESSIONS)
        )
    super().__init__(
        root_or_handler,
        listen=listen,
        port=port,
        writable=writable,
        create=create,
        overwrite=overwrite,
        timeout=timeout,
        retries=retries,
        options=options,
        max_sessions=max_sessions,
        reply_from_request_address=reply_from_request_address,
        dally=dally,
        on_complete=on_complete,
        limits=limits,
        ignore_broadcast=ignore_broadcast,
        backoff=backoff,
        max_timeout=max_timeout,
        trace=trace,
        port_range=port_range,
    )
    self._init_service()
    self._ready: "collections.deque[Session]" = collections.deque()
    self._pending_opens: "collections.deque[Tuple[Session, Any]]" = collections.deque()
    if open_in_thread is None:
        open_in_thread = not getattr(self.handler, "opens_fast", False)
    self._worker_count = max(1, workers) if open_in_thread else 0
    self._workers: Optional[concurrent.futures.ThreadPoolExecutor] = None
    self._selector: Any = None
    self._wake_r: Any = None
    self._wake_w: Any = None
    self._timers: List[Tuple[float, int, Session]] = []
    self._seq = 0
    self._buf = bytearray(_RECV_SIZE)
    self._view = memoryview(self._buf)

TFTPServerLimits(*, max_request_size=1024, max_filename_length=512, max_options=16, max_option_length=255, max_sessions_per_client=None, max_duration=None, max_idle=60.0)

Bounds on what one request, one client and one transfer may use.

:param max_request_size: bytes in an RRQ/WRQ datagram. RFC 2347 keeps a request within 512 octets; the default allows twice that. :param max_filename_length: characters in a requested filename. :param max_options: options in one request. :param max_option_length: characters in one option name or value. :param max_sessions_per_client: concurrent transfers from one client address (any port); None is unlimited. :param max_duration: seconds one transfer may take; None is unlimited. :param max_idle: seconds a transfer may go with no datagram from its peer before it ends; a timeout the client negotiated cannot extend it. Time the server spends waiting on its own handler does not count. None is unbounded.

A request over a limit is answered with ERROR 4 (ERROR 0 "server busy" for the per-client cap). Window memory is bounded separately, by :class:TFTPServerOptions max_window_bytes.

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

def __init__(
    self,
    *,
    max_request_size: int = 1024,
    max_filename_length: int = 512,
    max_options: int = 16,
    max_option_length: int = 255,
    max_sessions_per_client: Optional[int] = None,
    max_duration: Optional[float] = None,
    max_idle: Optional[float] = 60.0,
) -> None:
    if max_idle is not None and max_idle <= 0:
        raise ValueError("max_idle must be positive, or None for no bound")
    self.max_request_size = max_request_size
    self.max_filename_length = max_filename_length
    self.max_options = max_options
    self.max_option_length = max_option_length
    self.max_sessions_per_client = max_sessions_per_client
    self.max_duration = max_duration
    self.max_idle = max_idle

check(request)

Raise :class:TFTPError (4) if request exceeds a limit.

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

def check(self, request: RequestPacket) -> None:
    """Raise :class:`TFTPError` (4) if ``request`` exceeds a limit."""
    if len(request.filename) > self.max_filename_length:
        raise TFTPError(TFTPErrorCode.ILLEGAL_OPERATION, "filename too long")
    if len(request.options) > self.max_options:
        raise TFTPError(TFTPErrorCode.ILLEGAL_OPERATION, "too many options")
    for name, value in request.options.items():
        if len(name) > self.max_option_length or len(value) > self.max_option_length:
            raise TFTPError(TFTPErrorCode.ILLEGAL_OPERATION, "option too long")

TFTPStats(*names)

Monotonic counters, safe to read from any thread.

Server counters: requests (RRQ/WRQ seen on the listening port), refused (answered with an ERROR before a transfer started, or dropped by a limit), started, completed, failed, declined (the client refused the OACK before any data: a size probe), bytes_sent, bytes_received, retransmits. A relay counts requests, refused, started, completed, failed, bytes_to_clients, bytes_from_clients. snapshot() returns them all, plus active from the owner.

Source code in src/tftp/server/_stats.py

def __init__(self, *names: str) -> None:
    self._lock = threading.Lock()
    self._values: Dict[str, int] = {name: 0 for name in names}

TFTPWriter

Bases: Protocol

A binary sink a synchronous server writes an upload into.

Required: write(data) and close(). close is called once the last block is written and before it is acknowledged, so a failing close reaches the client as an ERROR; it may raise :class:tftp.WouldBlock while the data is still on its way elsewhere.

Optional members:

• abort() -- called instead of close when the transfer fails;
• copies_writes -- True when write copies what it is given. Without it write receives bytes, because the argument would otherwise be a view of a receive buffer that the next packet reuses (standard file objects always count as copying);
• set_wakeup(callback) -- as for :class:TFTPReader.

ThreadedHandler(handler, *, executor=None)

Gives a synchronous handler to :class:~tftp.AsyncTFTPServer.

open_read and open_write of handler run in executor (the loop's default one when None), so a handler that blocks never stalls the loop, unless the handler has opens_fast = True, when they run on the loop. The streams handler returns are synchronous and are read and written on the loop thread, as a synchronous server's are on its own.

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

def __init__(
    self, handler: TFTPHandler, *, executor: Optional[concurrent.futures.Executor] = None
) -> None:
    if has_coroutine_hooks(handler):
        raise TypeError(
            "ThreadedHandler wraps a synchronous handler; %r has coroutine hooks" % (handler,)
        )
    self.handler = handler
    self.executor = executor
    self.opens_fast = bool(getattr(handler, "opens_fast", False))