tftp.backends

tftp.backends

Request handlers beyond the local filesystem.

• :class:FilesystemBackend -- a directory (what TFTPServer("/srv/tftp") serves).
• :class:MemoryBackend -- a dictionary of files.
• :class:HTTPBackend -- a TFTP-to-HTTP(S) gateway (urllib, no extra dependency); PUT for uploads.
• :class:UpstreamBackend -- a terminating proxy to another TFTP server, each side negotiating its own options.
• :class:Remap -- rewrite the requested filenames, with regular expressions, before another handler sees them.
• :class:PerClient -- serve each client from a directory named for its address.
• :class:CaseInsensitive -- a :class:FilesystemBackend that finds files whatever the case of the requested name.
• :class:Pipe -- the bounded, backpressured byte pipe they use to feed a transfer from a worker thread; usable for your own handlers.

CaseInsensitive(root, *, writable=False, create=True, overwrite=False, backslash=True, max_upload=None)

Bases: FilesystemBackend

A :class:FilesystemBackend that matches names regardless of case.

The exact name wins when it exists; otherwise each component is looked up case-insensitively (firmware asking for \Boot\BCD finds boot/bcd). An upload's final component keeps the requested case. The corrected name goes through the normal containment checks, so nothing a client names leaves root. Takes the arguments of :class:FilesystemBackend.

Looking a name up lists each directory on the way, which costs a request more than the exact lookup does.

Source code in src/tftp/backends/_filesystem.py

def __init__(
    self,
    root: Union[str, "os.PathLike[str]"],
    *,
    writable: bool = False,
    create: bool = True,
    overwrite: bool = False,
    backslash: bool = True,
    max_upload: Optional[int] = None,
) -> None:
    self.root = os.path.realpath(os.fspath(root))
    if not os.path.isdir(self.root):
        raise NotADirectoryError(self.root)
    self.writable = writable
    self.create = create
    self.overwrite = overwrite
    self.backslash = backslash
    self.max_upload = max_upload

FilesystemBackend(root, *, writable=False, create=True, overwrite=False, backslash=True, max_upload=None)

Serves the files under root.

:param root: the directory served. Requests can never reach outside it: .. components are refused, and the resolved path (symlinks included) must still be inside root. :param writable: accept WRQ at all. :param create: a WRQ may create a file that does not exist. :param overwrite: a WRQ may replace a file that exists. :param backslash: treat \ in filenames as a separator, as Windows boot loaders send them (\boot\bcd). :param max_upload: refuse uploads announced (tsize) or grown larger than this many bytes, with ERROR 3.

Uploads go through :class:AtomicWriter, so a failed or partial upload leaves no trace, and directories are never created.

Source code in src/tftp/backends/_filesystem.py

def __init__(
    self,
    root: Union[str, "os.PathLike[str]"],
    *,
    writable: bool = False,
    create: bool = True,
    overwrite: bool = False,
    backslash: bool = True,
    max_upload: Optional[int] = None,
) -> None:
    self.root = os.path.realpath(os.fspath(root))
    if not os.path.isdir(self.root):
        raise NotADirectoryError(self.root)
    self.writable = writable
    self.create = create
    self.overwrite = overwrite
    self.backslash = backslash
    self.max_upload = max_upload

resolve(filename)

The local path for filename, or :class:TFTPError (2) if it escapes.

Source code in src/tftp/backends/_filesystem.py

def resolve(self, filename: str) -> str:
    """The local path for ``filename``, or :class:`TFTPError` (2) if it escapes."""
    name = filename.replace("\\", "/") if self.backslash else filename
    parts = []
    for part in name.split("/"):
        if part in ("", "."):
            continue
        if part == ".." or "\0" in part:
            raise TFTPError(TFTPErrorCode.ACCESS_VIOLATION)
        if _WINDOWS and _refused_on_windows(part):
            raise TFTPError(TFTPErrorCode.ACCESS_VIOLATION)
        parts.append(part)
    if not parts:
        raise TFTPError(TFTPErrorCode.FILE_NOT_FOUND)
    path = os.path.join(self.root, *parts)
    real = os.path.realpath(path)
    try:
        inside = os.path.commonpath([real, self.root]) == self.root
    except ValueError:  # different drives on Windows
        inside = False
    if not inside:
        raise TFTPError(TFTPErrorCode.ACCESS_VIOLATION)
    return real

HTTPBackend(base_url=None, *, url_for=None, writable=False, headers=None, timeout=10.0, buffer=1 << 20, opener=None)

Serve (and optionally accept, as PUT) files from an HTTP server.

:param base_url: prefix; the normalised filename is appended, percent-encoded (http://images/pxe/ + boot/x). Its scheme must be http or https. :param url_for: alternative to base_url: url_for(context) -> str, for routing by client, filename or anything else. A request whose URL is not http or https is refused with ERROR 2. :param writable: map WRQ to an HTTP PUT of the uploaded body. :param headers: sent with every request (Authorization...); they win over the default User-Agent. :param timeout: seconds for connecting and for each read. :param buffer: bytes buffered between HTTP and TFTP per transfer. :param opener: a urllib.request.OpenerDirector (proxies, TLS context). The default opener speaks http and https only and follows redirects between them; with an opener of your own the scheme of base_url and of url_for's URLs is yours to choose.

HTTP 404/410 become ERROR 1, 401/403 ERROR 2, 409 ERROR 6, 413/507 ERROR 3, anything else ERROR 0; a Content-Length answers tsize. The request is made in a server worker thread, never on the event loop.

A download is read ahead of the transfer by a few KiB until the peer has acknowledged data, and by up to buffer octets after. An upload's PUT carries the announced tsize as its Content-Length (chunked without one, and for netascii) and sends the last octets only when the transfer is complete: more octets than announced fail the transfer with ERROR 3, fewer with ERROR 0, and the origin stores nothing.

Source code in src/tftp/backends/_http.py

def __init__(
    self,
    base_url: Optional[str] = None,
    *,
    url_for: Optional[Callable[[TFTPRequestContext], str]] = None,
    writable: bool = False,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 10.0,
    buffer: int = 1 << 20,
    opener: Optional[urllib.request.OpenerDirector] = None,
) -> None:
    if (base_url is None) == (url_for is None):
        raise ValueError("give exactly one of base_url and url_for")
    if opener is None and base_url is not None and urlsplit(base_url).scheme.lower() not in _SCHEMES:
        raise ValueError("base_url is an http or https URL, not %r" % base_url)
    self.base_url = base_url
    self.url_for = url_for
    self.writable = writable
    self.headers = dict(headers or {})
    self.timeout = timeout
    self.buffer = buffer
    self._restricted = opener is None
    self._open = (opener or _http_opener()).open
    self._agent = _user_agent()
    self._refused_warned = False

MemoryBackend(files=None, *, writable=False, overwrite=True, max_upload=DEFAULT_MAX_UPLOAD, max_entries=DEFAULT_MAX_ENTRIES)

Serves files (name -> bytes); uploads land in it when writable.

Names are normalised (:func:normalize_name), so /boot/x and boot\x find the same entry. An upload replaces the entry only once it is complete. Safe to read and update files from other threads.

Whatever an anonymous peer may upload is bounded. max_upload is the octets one upload may hold (default 16 MiB): an announced tsize above it is refused with ERROR 3 at the request, and an upload that grows past it fails with ERROR 3. max_entries is the names the backend may hold (default 1024, those in files included): an upload to a new name beyond it fails with ERROR 3, while a name that exists can still be replaced. The memory a writable backend can hold is at most the product of the two; None removes a bound.

Source code in src/tftp/backends/_memory.py

def __init__(
    self,
    files: Optional[Mapping[str, bytes]] = None,
    *,
    writable: bool = False,
    overwrite: bool = True,
    max_upload: Optional[int] = DEFAULT_MAX_UPLOAD,
    max_entries: Optional[int] = DEFAULT_MAX_ENTRIES,
) -> None:
    self.files: Dict[str, bytes] = {normalize_name(k): bytes(v) for k, v in (files or {}).items()}
    self.writable = writable
    self.overwrite = overwrite
    self.max_upload = _bound("max_upload", max_upload)
    self.max_entries = _bound("max_entries", max_entries)
    self._lock = threading.Lock()

PerClient(root, make, *, fallback=True)

Serves root/<client address>/ to a client whose directory exists.

A client with no directory of its own is served root itself by default, so it can read every other client's directory: that is a convenience for a fleet that mostly shares files, not isolation. With fallback=False such a client is answered ERROR 1 (file not found) to every request, and a client reaches only its own directory (what it names stays inside that directory, as :class:FilesystemBackend keeps it).

The directory name is the client's address with each : written - (fe80--1, valid on Windows), and a v4-mapped IPv6 address is named by its IPv4 address (see :meth:directory).

:param root: the directory holding one directory per client. :param make: make(directory) returns the handler for a directory, for example lambda d: FilesystemBackend(d, writable=True); it is called once for each directory a client is served from. It should return a handler that opens quickly (a local directory): this one opens inline on the server's thread. :param fallback: serve root itself to a client with no directory.

Source code in src/tftp/backends/_per_client.py

def __init__(
    self,
    root: Union[str, "os.PathLike[str]"],
    make: Callable[[str], TFTPHandler],
    *,
    fallback: bool = True,
) -> None:
    self.root = os.path.realpath(os.fspath(root))
    self.make = make
    self.fallback = fallback
    self._handlers: Dict[str, TFTPHandler] = {}

directory(peer) staticmethod

The directory name for the client at peer ((host, port, ...)).

Source code in src/tftp/backends/_per_client.py

@staticmethod
def directory(peer: Tuple[Any, ...]) -> str:
    """The directory name for the client at ``peer`` (``(host, port, ...)``)."""
    from netimps import split_zone, unmap

    return str(unmap(split_zone(str(peer[0]))[0])).replace(":", "-")

handler_for(context)

The handler of the directory context's client is served from.

:raises TFTPError: ERROR 1, for a client with no directory when fallback is false.

Source code in src/tftp/backends/_per_client.py

def handler_for(self, context: TFTPRequestContext) -> TFTPHandler:
    """The handler of the directory ``context``'s client is served from.

    :raises TFTPError: ERROR 1, for a client with no directory when
        ``fallback`` is false.
    """
    directory = os.path.join(self.root, self.directory(context.peer))
    if not os.path.isdir(directory):
        if not self.fallback:
            raise TFTPError(TFTPErrorCode.FILE_NOT_FOUND)
        directory = self.root
    handler = self._handlers.get(directory)
    if handler is None:
        handler = self._handlers[directory] = self.make(directory)
    return handler

Pipe(capacity=1 << 20, size=None)

A bounded FIFO of bytes with one producer and one consumer.

:param capacity: bytes buffered at most; a producer blocks (or gets WouldBlock) beyond it. This is the backpressure that keeps a slow consumer from making a fast producer buffer a whole file. :param size: the total size, when the producer knows it in advance; a server answers tsize from it.

Source code in src/tftp/backends/_pipe.py

def __init__(self, capacity: int = 1 << 20, size: Optional[int] = None) -> None:
    if capacity < 1:
        raise ValueError("capacity must be positive")
    self.capacity = capacity
    self.size = size
    self._buffer = bytearray()
    self._lock = threading.Condition()
    self._eof = False  # producer finished
    self._closed = False  # consumer finished (or gave up)
    self._error: Optional[BaseException] = None
    self._wakeup: Optional[Callable[[], None]] = None
    #: Upload mode: set once the worker has reported through set_result().
    self._done = threading.Event()
    self._result_error: Optional[BaseException] = None

abort(error=None)

Give up from the transfer side; the worker's next call raises.

Bytes still buffered are dropped: nothing will read them.

Source code in src/tftp/backends/_pipe.py

def abort(self, error: Optional[BaseException] = None) -> None:
    """Give up from the transfer side; the worker's next call raises.

    Bytes still buffered are dropped: nothing will read them.
    """
    with self._lock:
        self._closed = True
        self._eof = True
        del self._buffer[:]
        if self._error is None:
            self._error = error or TFTPError(0, "transfer aborted")
        self._lock.notify_all()

close()

The transfer side is done.

Download pipe (the transfer reads): the producer's next put raises, which stops a worker that is still fetching.

Upload pipe (:meth:for_upload, the transfer writes): marks the end of the data, then raises :class:WouldBlock until the worker calls :meth:set_result -- so the client's final ACK waits until the data really reached its destination -- and re-raises the worker's error if it had one.

Source code in src/tftp/backends/_pipe.py

def close(self) -> None:
    """The transfer side is done.

    Download pipe (the transfer reads): the producer's next ``put``
    raises, which stops a worker that is still fetching.

    Upload pipe (:meth:`for_upload`, the transfer writes): marks the end
    of the data, then raises :class:`WouldBlock` until the worker calls
    :meth:`set_result` -- so the client's final ACK waits until the data
    really reached its destination -- and re-raises the worker's error if
    it had one.
    """
    if self._upload:
        with self._lock:
            self._eof = True
            self._lock.notify_all()
        if not self._done.is_set():
            raise WouldBlock()
        if self._result_error is not None:
            raise self._result_error
        return
    with self._lock:
        self._closed = True
        self._lock.notify_all()

finish(error=None)

The producer is done (or failed with error).

Source code in src/tftp/backends/_pipe.py

def finish(self, error: Optional[BaseException] = None) -> None:
    """The producer is done (or failed with ``error``)."""
    with self._lock:
        self._eof = True
        if error is not None and self._error is None:
            self._error = error
        self._lock.notify_all()
    self._wake()

for_upload()

Make this an upload pipe, whose close() waits for :meth:set_result.

Source code in src/tftp/backends/_pipe.py

def for_upload(self) -> "Pipe":
    """Make this an upload pipe, whose close() waits for :meth:`set_result`."""
    self._upload = True
    return self

get(n=65536, timeout=None)

Blocking read for a worker thread; b"" at the end.

Source code in src/tftp/backends/_pipe.py

def get(self, n: int = 65536, timeout: Optional[float] = None) -> bytes:
    """Blocking read for a worker thread; ``b""`` at the end."""
    with self._lock:
        if not self._lock.wait_for(lambda: self._buffer or self._eof or self._error, timeout):
            raise TimeoutError("pipe producer stalled")
        if self._buffer:
            chunk = bytes(self._buffer[:n])
            del self._buffer[:n]
            self._lock.notify_all()
        elif self._error is not None:
            raise self._error
        else:
            chunk = b""
    if chunk:
        self._wake()
    return chunk

put(data, timeout=None)

Blocking write for a worker thread; raises if the consumer is gone.

Source code in src/tftp/backends/_pipe.py

def put(self, data: Union[bytes, bytearray, memoryview], timeout: Optional[float] = None) -> None:
    """Blocking write for a worker thread; raises if the consumer is gone."""
    view = memoryview(data)
    while view:
        with self._lock:
            if not self._lock.wait_for(
                lambda: self._closed or len(self._buffer) < self.capacity, timeout
            ):
                raise TimeoutError("pipe consumer stalled")
            if self._closed:
                raise BrokenPipeError("consumer closed the pipe")
            room = self.capacity - len(self._buffer)
            self._buffer += view[:room]
            view = view[room:]
            self._lock.notify_all()
        self._wake()

read(n=65536)

Blocking read so a worker thread can treat the pipe as a file.

Source code in src/tftp/backends/_pipe.py

def read(self, n: int = 65536) -> bytes:
    """Blocking ``read`` so a worker thread can treat the pipe as a file."""
    return self.get(n)

readinto(view)

Non-blocking read for a transfer: some bytes, 0 at the end, or WouldBlock.

Source code in src/tftp/backends/_pipe.py

def readinto(self, view: Union[bytearray, memoryview]) -> int:
    """Non-blocking read for a transfer: some bytes, 0 at the end, or WouldBlock."""
    with self._lock:
        if self._buffer:
            n = min(len(view), len(self._buffer))
            view[:n] = self._buffer[:n]
            del self._buffer[:n]
            self._lock.notify_all()
        elif self._error is not None:
            raise self._error
        elif self._eof:
            return 0
        else:
            raise WouldBlock()
    self._wake()
    return n

set_result(error=None)

Worker side of an upload: finished, successfully or with error.

Source code in src/tftp/backends/_pipe.py

def set_result(self, error: Optional[BaseException] = None) -> None:
    """Worker side of an upload: finished, successfully or with ``error``."""
    self._result_error = error
    self._done.set()
    self._wake()

set_wakeup(callback)

Called (from any thread) when the non-blocking side can proceed.

Source code in src/tftp/backends/_pipe.py

def set_wakeup(self, callback: Callable[[], None]) -> None:
    """Called (from any thread) when the non-blocking side can proceed."""
    self._wakeup = callback

write(data)

Non-blocking write for a transfer: all of data or WouldBlock.

Source code in src/tftp/backends/_pipe.py

def write(self, data: Union[bytes, bytearray, memoryview]) -> int:
    """Non-blocking write for a transfer: all of ``data`` or WouldBlock."""
    with self._lock:
        if self._error is not None:
            raise self._error
        if self._closed:
            raise BrokenPipeError("consumer closed the pipe")
        if self._buffer and len(self._buffer) + len(data) > self.capacity:
            raise WouldBlock()
        self._buffer += data
        self._lock.notify_all()
    return len(data)

Remap(inner, rules)

Rewrites each requested filename with the first rule that matches it, then asks inner.

Only the first rule whose pattern is found in the name applies, and it replaces every match in it (:meth:re.Pattern.sub, so \1 names a group). inner sees the rewritten name and decides what it may reach: a rewrite cannot take a request out of the directory a :class:FilesystemBackend serves. Everything else about the request (the listing flag, the arrival interface) is carried over.

:param inner: the handler that serves the rewritten names. :param rules: "REGEX=REPLACEMENT" text (split at the first =) or (pattern, replacement) pairs, tried in order. :raises TFTPValueError: a rule without = or a regular expression that does not compile. :raises TypeError: a rule of another shape.

Source code in src/tftp/backends/_remap.py

def __init__(self, inner: TFTPHandler, rules: Iterable[_RuleLike]) -> None:
    self.inner = inner
    self.rules: Tuple[Tuple["Pattern[str]", str], ...] = tuple(_rule(rule) for rule in rules)
    self.opens_fast = getattr(inner, "opens_fast", False)

rewrite(filename)

filename as the first matching rule rewrites it, or unchanged.

Source code in src/tftp/backends/_remap.py

def rewrite(self, filename: str) -> str:
    """``filename`` as the first matching rule rewrites it, or unchanged."""
    for pattern, replacement in self.rules:
        if pattern.search(filename):
            return pattern.sub(replacement, filename)
    return filename

UpstreamBackend(upstream, *, client_options=None, buffer=1 << 20, stall_timeout=30.0, writable=False)

Serve every request from an upstream TFTP server.

:param upstream: "host", "host:port", ("host", port), or a callable upstream(context) returning one of those -- route by client address, filename, the interface the request arrived on. :param client_options: keyword arguments for the upstream :class:tftp.TFTPClient (blksize, windowsize, timeout...). :param buffer: bytes buffered between the two transfers. :param stall_timeout: seconds one side may wait on the other before the transfer is abandoned. :param writable: forward WRQ upstream (otherwise ERROR 2).

The upstream request is made in a server worker thread. An RRQ is answered only once the upstream has answered: its errors (file not found...) reach the client with the same code and text, and its tsize is passed through. A WRQ's final ACK is sent only once the upstream has acknowledged the whole file.

Source code in src/tftp/backends/_proxy.py

def __init__(
    self,
    upstream: Union[_Target, Callable[[TFTPRequestContext], _Target]],
    *,
    client_options: Optional[Mapping[str, Any]] = None,
    buffer: int = 1 << 20,
    stall_timeout: float = 30.0,
    writable: bool = False,
) -> None:
    self.upstream = upstream
    self.client_options = dict(client_options or {})
    self.buffer = buffer
    self.stall_timeout = stall_timeout
    self.writable = writable

normalize_name(filename, backslash=True)

/a\b//c -> a/b/c: the key a request maps to.

Source code in src/tftp/backends/_memory.py

def normalize_name(filename: str, backslash: bool = True) -> str:
    """``/a\\b//c`` -> ``a/b/c``: the key a request maps to."""
    name = filename.replace("\\", "/") if backslash else filename
    return "/".join(part for part in name.split("/") if part and part != ".")