Exceptions and results

tftp.exceptions

Every exception the package raises on its own account, and the mapping from OS errors to TFTP error codes.

:class:TFTPError is the one base. It plays three roles, told apart by the subclass:

• what a server handler raises to refuse a request: the code and message become the ERROR packet the client receives (:class:TFTPError itself, or :class:TFTPProtocolError);
• what a client raises for an ERROR the server sent: :class:RemoteError and its leaves, one per defined error code;
• what a transfer that failed reports in TransferResult.error, whichever of these it was, including :class:TransferTimeoutError, :class:TransferAbortedError and :class:TransferTooLargeError for a local deadline, an abort() or a client's max_size.

Text that is not the value it was asked to become raises :class:TFTPValueError, which is also a :class:ValueError: :class:TFTPDecodeError for bytes that are not a packet. A caller's own mistake, a wrong argument type or an option out of range, is plain :class:TypeError or :class:ValueError, not one of these. A file that is not a capture raises pktcap.CaptureFormatError and a filter expression that does not compile pktcap.CaptureFilterError, each a :class:ValueError of that library.

:class:WouldBlock is a signal that a source or sink has nothing ready, not a failure, so it is a :class:BlockingIOError and not a :class:TFTPError.

Every class is rebuilt by type(*args), so each one copies and pickles.

AccessViolation(code=None, message='')

Bases: RemoteError

ERROR 2.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

DiskFull(code=None, message='')

Bases: RemoteError

ERROR 3: disk full or allocation exceeded.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

FileAlreadyExists(code=None, message='')

Bases: RemoteError

ERROR 6.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

FileNotFound(code=None, message='')

Bases: RemoteError

ERROR 1.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

IllegalOperation(code=None, message='')

Bases: RemoteError

ERROR 4.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

NoSuchUser(code=None, message='')

Bases: RemoteError

ERROR 7.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

OptionNegotiationError(code=None, message='')

Bases: RemoteError

ERROR 8: the peer refused option negotiation.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

RemoteError(code=None, message='')

Bases: TFTPError

The peer sent an ERROR packet.

Raised as the subclass for its code (:class:FileNotFound, :class:AccessViolation, ...) so callers can catch by kind; code and message hold exactly what the peer sent. The leaves describe the server's file, so none is also a FileNotFoundError or PermissionError: :mod:tftp.path translates where a path caller expects the builtin.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

from_code(code, message='') classmethod

The :class:RemoteError subclass instance for code (cls itself for a code with no leaf).

Source code in src/tftp/exceptions.py

@classmethod
def from_code(cls, code: int, message: str = "") -> "RemoteError":
    """The :class:`RemoteError` subclass instance for ``code`` (``cls`` itself for a code with no leaf)."""
    return _REMOTE_CLASSES.get(code, cls)(code, message)

TFTPDecodeError(message='')

Bases: TFTPValueError

The bytes do not form a valid TFTP packet.

Carries ERROR 4 (illegal operation), the code a peer is answered with.

Source code in src/tftp/exceptions.py

def __init__(self, message: str = "") -> None:
    super().__init__(self._CODE, message)

TFTPError(code=TFTPErrorCode.NOT_DEFINED, message='')

Bases: Exception

A TFTP failure carrying the ERROR code that describes it.

A server handler raises this to refuse a request: the code and message become the ERROR packet the client receives.

:param code: an int in 0..65535 (unknown codes are kept: the wire carries any 16-bit value). TypeError for anything else, which catches TFTPError("message"); ValueError outside the range. :param message: text; TypeError for anything else. A NUL is replaced and a long text cut when the ERROR is encoded.

Source code in src/tftp/exceptions.py

def __init__(self, code: int = TFTPErrorCode.NOT_DEFINED, message: str = "") -> None:
    if isinstance(code, bool) or not isinstance(code, int):
        raise TypeError(
            "an error code is an int, not %s (the message is the second argument)" % type(code).__name__
        )
    if not 0 <= code <= 65535:
        raise ValueError("an error code is 0 to 65535, not %d" % code)
    if not isinstance(message, str):
        raise TypeError("an error message is text, not %s" % type(message).__name__)
    self.code = TFTPErrorCode(code)
    self.message = message or _default_message(code)
    super().__init__(*self._constructor_args())

from_exception(exc) classmethod

The ERROR to send for exc, always a plain :class:TFTPError or exc itself.

A :class:TFTPError is used as is. An OSError maps by errno (ENOENT -> file not found, EACCES -> access violation, ENOSPC -> disk full, ...). Its message is the generic one for the code, never the OS text, which would disclose server paths.

Source code in src/tftp/exceptions.py

@classmethod
def from_exception(cls, exc: BaseException) -> "TFTPError":
    """The ERROR to send for ``exc``, always a plain :class:`TFTPError` or ``exc`` itself.

    A :class:`TFTPError` is used as is. An ``OSError`` maps by errno
    (``ENOENT`` -> file not found, ``EACCES`` -> access violation,
    ``ENOSPC`` -> disk full, ...). Its message is the generic one for the
    code, never the OS text, which would disclose server paths.
    """
    if isinstance(exc, TFTPError):
        return exc
    if isinstance(exc, OSError):
        code = _ERRNO_CODES.get(exc.errno or 0)
        if code is None:
            if isinstance(exc, FileNotFoundError):
                code = TFTPErrorCode.FILE_NOT_FOUND
            elif isinstance(exc, PermissionError):
                code = TFTPErrorCode.ACCESS_VIOLATION
            elif isinstance(exc, FileExistsError):
                code = TFTPErrorCode.FILE_EXISTS
            else:
                code = TFTPErrorCode.NOT_DEFINED
        return TFTPError(code)
    return TFTPError(TFTPErrorCode.NOT_DEFINED)

TFTPProtocolError(message, code=TFTPErrorCode.ILLEGAL_OPERATION)

Bases: TFTPError

The peer broke the protocol (malformed packet, bad option ack...).

Source code in src/tftp/exceptions.py

def __init__(self, message: str, code: int = TFTPErrorCode.ILLEGAL_OPERATION) -> None:
    super().__init__(code, message)

TFTPValueError(message='')

Bases: TFTPError, ValueError

Text that is not the value it was asked to become: a malformed tftp:// URL, for one.

Also a :class:ValueError, so except ValueError keeps catching it.

Source code in src/tftp/exceptions.py

def __init__(self, message: str = "") -> None:
    super().__init__(self._CODE, message)

TransferAbortedError(message='transfer aborted')

Bases: TFTPError

Cancelled locally (abort(), a server shutting down).

Source code in src/tftp/exceptions.py

def __init__(self, message: str = "transfer aborted") -> None:
    super().__init__(TFTPErrorCode.NOT_DEFINED, message)

TransferTimeoutError(message='timed out')

Bases: TFTPError, TimeoutError

The peer stopped answering, or the transfer ran out of time.

Also a :class:TimeoutError; errno is None, since no OS call failed.

Source code in src/tftp/exceptions.py

def __init__(self, message: str = "timed out") -> None:
    super().__init__(TFTPErrorCode.NOT_DEFINED, message)

TransferTooLargeError(message='transfer too large')

Bases: TFTPError

A download would bring more than the client's max_size.

Raised, and sent to the server as ERROR 3, before the octet that passes the bound is written; code is DISK_FULL.

Source code in src/tftp/exceptions.py

def __init__(self, message: str = "transfer too large") -> None:
    super().__init__(TFTPErrorCode.DISK_FULL, message)

UnknownTransferID(code=None, message='')

Bases: RemoteError

ERROR 5.

Source code in src/tftp/exceptions.py

def __init__(self, code: Optional[int] = None, message: str = "") -> None:
    super().__init__(self._CODE if code is None else code, message)

WouldBlock

Bases: BlockingIOError

Raised by a source or sink that has nothing ready yet.

A sender whose read raises it stops after the blocks it has, and a receiver whose write (or complete) raises it holds the block unacknowledged; both carry on when the driver calls Transfer.resume. This is how a slow backend (an upstream server, an HTTP fetch) applies backpressure without blocking the loop that serves every other transfer.

tftp.TransferResult(filename, operation, mode, peer, local, bytes, blocks, retransmits, duration, negotiated, error=None)

The outcome of one transfer, on either side.

:ivar filename: as requested. :ivar operation: "read" for an RRQ, "write" for a WRQ, whichever side reports it. :ivar mode: "octet" or "netascii". :ivar peer: the other side's (host, port, ...) transfer address. :ivar local: this side's transfer address. :ivar bytes: payload bytes moved (netascii-encoded size on the wire). :ivar blocks: DATA blocks moved, excluding retransmissions. :ivar retransmits: packets sent again after a timeout or a lost block. :ivar duration: seconds from request to completion. :ivar negotiated: the parameters the transfer ran with. :ivar error: the failure, or None on success.

Source code in src/tftp/_result.py

def __init__(
    self,
    filename: str,
    operation: str,
    mode: str,
    peer: Tuple[Any, ...],
    local: Tuple[Any, ...],
    bytes: int,
    blocks: int,
    retransmits: int,
    duration: float,
    negotiated: Negotiated,
    error: Optional[TFTPError] = None,
) -> None:
    self.filename = filename
    self.operation = operation
    self.mode = mode
    self.peer = peer
    self.local = local
    self.bytes = bytes
    self.blocks = blocks
    self.retransmits = retransmits
    self.duration = duration
    self.negotiated = negotiated
    self.error = error

throughput property

Payload bytes per second.

to_dict()

The result as a dictionary of JSON types, one object per transfer.

ok is :attr:is_ok; peer is [host, port]; duration is seconds to the microsecond; blksize, windowsize, tsize and options come from :attr:negotiated; error is the failure's text, or None.

Source code in src/tftp/_result.py

def to_dict(self) -> Dict[str, Any]:
    """The result as a dictionary of JSON types, one object per transfer.

    ``ok`` is :attr:`is_ok`; ``peer`` is ``[host, port]``; ``duration`` is
    seconds to the microsecond; ``blksize``, ``windowsize``, ``tsize`` and
    ``options`` come from :attr:`negotiated`; ``error`` is the failure's
    text, or ``None``.
    """
    negotiated = self.negotiated
    return {
        "ok": self.is_ok,
        "operation": self.operation,
        "filename": self.filename,
        "mode": self.mode,
        "peer": list(self.peer[:2]),
        "bytes": self.bytes,
        "blocks": self.blocks,
        "retransmits": self.retransmits,
        "duration": round(self.duration, 6),
        "blksize": negotiated.blksize,
        "windowsize": negotiated.windowsize,
        "tsize": negotiated.tsize,
        "options": dict(negotiated.options),
        "error": None if self.error is None else str(self.error),
    }