tftp.relay

tftp.relay

Transparent TFTP relay with pluggable routing.

See :class:TFTPRelay and its asyncio twin :class:AsyncTFTPRelay. For a proxy that terminates both sessions (each side with its own options) use :class:tftp.backends.UpstreamBackend with a TFTPServer instead.

AsyncTFTPRelay(route, *, listen=None, port=69, idle_timeout=IDLE_TIMEOUT, max_duration=MAX_DURATION, linger=LINGER, upstream_src=None, limits=None, max_sessions=DEFAULT_MAX_SESSIONS, ignore_broadcast=True, reply_from_request_address=True, trace=None, on_session_end=None, port_range=None)

Bases: AsyncioService, RelayBase

:class:tftp.relay.TFTPRelay for asyncio; the same arguments, but route is a coroutine.

route is an upstream (as for TFTPRelay) or an async def route(request, context) -> upstream | None; a plain function is a TypeError. The route and the lookup of an upstream's name (in the loop's executor, unless it is an address) are bounded together by idle_timeout and count against max_sessions; one that outlasts it gets the client ERROR 0 "relay error". Use as async with AsyncTFTPRelay(...) as relay: await relay.serve_forever(), or await relay.start() ... await relay.aclose().

The lifecycle is :class:tftp.AsyncTFTPServer's: bind() is plain, shutdown() never blocks and is safe from any thread, await wait_closed() returns once serving has stopped and await aclose() is final. Stopping ends every transfer in flight, both ends getting ERROR 0 "relay shutting down".

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

def __init__(
    self,
    route: Union[AsyncRouteFunction, UpstreamLike],
    *,
    listen: ListenLike = None,
    port: int = 69,
    idle_timeout: float = IDLE_TIMEOUT,
    max_duration: float = MAX_DURATION,
    linger: float = LINGER,
    upstream_src: Optional[HostLike] = None,
    limits: Optional[TFTPServerLimits] = None,
    max_sessions: Optional[int] = DEFAULT_MAX_SESSIONS,
    ignore_broadcast: bool = True,
    reply_from_request_address: bool = True,
    trace: Optional[Callable[[PacketEvent], Any]] = None,
    on_session_end: Optional[Callable[[RelaySummary], Any]] = None,
    port_range: Optional[PortRangeLike] = None,
) -> None:
    super().__init__(
        route,
        listen=listen,
        port=port,
        idle_timeout=idle_timeout,
        max_duration=max_duration,
        linger=linger,
        upstream_src=upstream_src,
        limits=limits,
        max_sessions=max_sessions,
        ignore_broadcast=ignore_broadcast,
        reply_from_request_address=reply_from_request_address,
        trace=trace,
        on_session_end=on_session_end,
        port_range=port_range,
    )
    self._tasks: Set["asyncio.Future[None]"] = set()
    self._timer: Optional[asyncio.TimerHandle] = None
    self._ticking = False
    self._abandoning = False
    self._init_service()

RelaySummary

Bases: NamedTuple

What :class:TFTPRelay reports when a relayed transfer ends.

reason is "complete", "error" (an ERROR passed through, see error), "idle", "lifetime" or "shutdown".

to_dict()

The summary as a dictionary of JSON types: client and upstream are [host, port].

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

def to_dict(self) -> Dict[str, Any]:
    """The summary as a dictionary of JSON types: ``client`` and ``upstream`` are ``[host, port]``."""
    record = self._asdict()
    record["client"] = list(self.client[:2])
    record["upstream"] = list(self.upstream[:2])
    return record

RouteTable(routes, default=None)

Try routes in order; the first that returns an upstream wins.

default (optional) answers when none does.

Source code in src/tftp/relay/_routing.py

def __init__(self, routes: Iterable[RouteFunction], default: Optional[UpstreamLike] = None) -> None:
    self.routes = list(routes)
    self.default = Upstream.parse(default) if default is not None else None

TFTPRelay(route, *, listen=None, port=69, idle_timeout=IDLE_TIMEOUT, max_duration=MAX_DURATION, linger=LINGER, upstream_src=None, limits=None, max_sessions=DEFAULT_MAX_SESSIONS, ignore_broadcast=True, reply_from_request_address=True, trace=None, on_session_end=None, port_range=None)

Bases: SelectorService, RelayBase

Forward TFTP requests to upstream servers, transparently.

:param route: an upstream ("host", "host:port", (host, port), :class:Upstream) for every request, or a route callable route(request, context) -> upstream | None (see :mod:tftp.relay). None refuses with ERROR 2. The callable is a plain function: a coroutine function is a TypeError, which :class:AsyncTFTPRelay takes. :param listen, port: where to listen for requests (as for TFTPServer). :param idle_timeout: end a transfer after this long without traffic. Keep it above the largest timeout a peer may negotiate times its retries. :param max_duration: end any transfer after this long. :param linger: after a recognised final DATA/ACK exchange (or an ERROR, up to 1 s), keep forwarding this long for retransmissions. :param upstream_src: local address to send upstream from (a string, an ipaddress address or a netimps.Host). :param max_sessions: concurrent relayed transfers; requests beyond it get ERROR 0 "relay busy". The default is 250 on every platform (two sockets each); None is unlimited, except on Windows, where it is 250 and a number above 255 is a ValueError (select()'s limit). :param limits, ignore_broadcast, reply_from_request_address: as for TFTPServer. :param trace: trace(PacketEvent) for every datagram received or sent, with role="relay" and leg "client"/"upstream". :param on_session_end: on_session_end(RelaySummary) when a relayed transfer ends. :param port_range: ports for both sockets of each relayed transfer (the client side and the upstream side), as for TFTPServer.

The constructor opens nothing: :meth:bind does, and :meth:start, :meth:serve_forever and with call it. The lifecycle is :class:TFTPServer's: shutdown() never blocks, wait_closed() returns once serving has stopped, close() is final.

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

def __init__(
    self,
    route: Union[RouteFunction, UpstreamLike],
    *,
    listen: ListenLike = None,
    port: int = 69,
    idle_timeout: float = IDLE_TIMEOUT,
    max_duration: float = MAX_DURATION,
    linger: float = LINGER,
    upstream_src: Optional[HostLike] = None,
    limits: Optional[TFTPServerLimits] = None,
    max_sessions: Optional[int] = DEFAULT_MAX_SESSIONS,
    ignore_broadcast: bool = True,
    reply_from_request_address: bool = True,
    trace: Optional[Callable[[PacketEvent], Any]] = None,
    on_session_end: Optional[Callable[[RelaySummary], Any]] = None,
    port_range: Optional[PortRangeLike] = None,
) -> None:
    super().__init__(
        route,
        listen=listen,
        port=port,
        idle_timeout=idle_timeout,
        max_duration=max_duration,
        linger=linger,
        upstream_src=upstream_src,
        limits=limits,
        max_sessions=max_sessions,
        ignore_broadcast=ignore_broadcast,
        reply_from_request_address=reply_from_request_address,
        trace=trace,
        on_session_end=on_session_end,
        port_range=port_range,
    )
    self._selector: Any = None
    self._wake_r: Any = None
    self._wake_w: Any = None
    self._buf = bytearray(_RECV)
    self._init_service()

Upstream(host, port=69) dataclass

Where to forward: an upstream server's host and request port.

An immutable value that equals only another Upstream. host is whatever was given: a name or address string, an ipaddress address, or a netimps.Host; port is an int in 1..65535 (TypeError for another type, :class:TFTPValueError outside the range).

parse(value) classmethod

An :class:Upstream from "host", "host:port", "[v6]:port", an address or netimps.Host (port 69 unless its text has one), (host, port) or an :class:Upstream.

:raises TypeError: a host that is none of those. :raises ValueError: host text netimps cannot read, or a bad port.

Source code in src/tftp/relay/_routing.py

@classmethod
def parse(cls, value: "UpstreamLike") -> "Upstream":
    """An :class:`Upstream` from ``"host"``, ``"host:port"``, ``"[v6]:port"``,
    an address or ``netimps.Host`` (port 69 unless its text has one),
    ``(host, port)`` or an :class:`Upstream`.

    :raises TypeError: a host that is none of those.
    :raises ValueError: host text netimps cannot read, or a bad port.
    """
    if isinstance(value, Upstream):
        return value if isinstance(value, cls) else cls(value.host, value.port)
    if isinstance(value, tuple):
        return cls(value[0], value[1])
    from netimps import split_host

    host, port = split_host(value, default_port=69)
    return cls(host, port or 69)

__getattr__(name)

Bind :class:AsyncTFTPRelay on first use, so importing the package does not import asyncio.

Source code in src/tftp/relay/__init__.py

def __getattr__(name: str) -> Any:
    """Bind :class:`AsyncTFTPRelay` on first use, so importing the package does not import asyncio."""
    if name == "AsyncTFTPRelay":
        from ._asyncio import AsyncTFTPRelay

        globals()[name] = AsyncTFTPRelay
        return AsyncTFTPRelay
    raise AttributeError("module %r has no attribute %r" % (__name__, name))

by_interface(table)

Route by arrival: an interface index (int), a netimps.Interface, or the local address the request was sent to (an address string or object, an ipaddress interface, a netimps.Host).

Needs pktinfo; without it neither is known and nothing matches.

Source code in src/tftp/relay/_routing.py

def by_interface(table: Dict[Union[int, Interface, HostLike], UpstreamLike]) -> RouteFunction:
    """Route by arrival: an interface index (``int``), a ``netimps.Interface``,
    or the local address the request was sent to (an address string or
    object, an ``ipaddress`` interface, a ``netimps.Host``).

    Needs pktinfo; without it neither is known and nothing matches.
    """
    from netimps import Host, Interface, split_zone, unmap

    by_index: Dict[int, Upstream] = {}
    by_address: Dict[Any, Upstream] = {}
    for key, target in table.items():
        if isinstance(key, bool):
            raise TypeError("by_interface keys are indexes, interfaces or addresses, not bool")
        if isinstance(key, int):
            by_index[key] = Upstream.parse(target)
        elif isinstance(key, Interface):
            by_index[key.index] = Upstream.parse(target)
        else:
            address = Host(key).ip()  # an ipaddress interface counts as its address
            if address is None:
                raise ValueError("by_interface: cannot resolve %r" % (key,))
            by_address[unmap(address)] = Upstream.parse(target)

    def route(request: RequestPacket, context: TFTPRequestContext) -> Optional[Upstream]:
        if context.interface_index and context.interface_index in by_index:
            return by_index[context.interface_index]
        if context.local_address is not None and by_address:
            return by_address.get(unmap(split_zone(context.local_address)[0]))
        return None

    return route

by_prefix(table)

Route by filename prefix: {"windows/": "wds.lan", "": "default.lan"}.

The longest matching prefix wins; leading / and \ are ignored and \ counts as /. A sequence may hold "PREFIX=HOST[:PORT]" text in place of a pair.

:raises TFTPValueError: text without = or without a target.

Source code in src/tftp/relay/_routing.py

def by_prefix(
    table: Union[Dict[str, UpstreamLike], Sequence[Union[str, Tuple[str, UpstreamLike]]]],
) -> RouteFunction:
    """Route by filename prefix: ``{"windows/": "wds.lan", "": "default.lan"}``.

    The longest matching prefix wins; leading ``/`` and ``\\`` are ignored
    and ``\\`` counts as ``/``. A sequence may hold ``"PREFIX=HOST[:PORT]"``
    text in place of a pair.

    :raises TFTPValueError: text without ``=`` or without a target.
    """
    prefixes = sorted(
        ((p.replace("\\", "/").lstrip("/"), Upstream.parse(t)) for p, t in _items(table)),
        key=lambda x: -len(x[0]),
    )

    def route(request: RequestPacket, context: TFTPRequestContext) -> Optional[Upstream]:
        name = request.filename.replace("\\", "/").lstrip("/")
        for prefix, target in prefixes:
            if name.startswith(prefix):
                return target
        return None

    return route

by_subnet(table)

Route by the client's address: {"10.1.0.0/16": "10.1.0.5", ...}.

Keys are anything netimps.parse(..., IPNetwork) takes: CIDR strings, ipaddress networks, interfaces (their network) or addresses (a /32 or /128). The most specific (longest prefix) matching network wins. A v4 client seen as ::ffff:a.b.c.d matches v4 networks. A sequence may hold "CIDR=HOST[:PORT]" text in place of a pair.

:raises TFTPValueError: text without = or without a target.

Source code in src/tftp/relay/_routing.py

def by_subnet(
    table: Union[Dict[IPNetworkLike, UpstreamLike], Sequence[Union[str, Tuple[IPNetworkLike, UpstreamLike]]]],
) -> RouteFunction:
    """Route by the client's address: ``{"10.1.0.0/16": "10.1.0.5", ...}``.

    Keys are anything ``netimps.parse(..., IPNetwork)`` takes: CIDR strings,
    ``ipaddress`` networks, interfaces (their network) or addresses (a /32 or
    /128). The most specific (longest prefix) matching network wins. A v4
    client seen as ``::ffff:a.b.c.d`` matches v4 networks. A sequence may hold
    ``"CIDR=HOST[:PORT]"`` text in place of a pair.

    :raises TFTPValueError: text without ``=`` or without a target.
    """
    from netimps import IPNetwork, parse

    networks = sorted(
        ((parse(net, IPNetwork), Upstream.parse(target)) for net, target in _items(table)),
        key=lambda pair: pair[0].prefixlen,
        reverse=True,
    )

    def route(request: RequestPacket, context: TFTPRequestContext) -> Optional[Upstream]:
        from netimps import unmap

        address = unmap(context.peer[0])
        for network, target in networks:
            if address.version == network.version and address in network:
                return target
        return None

    return route