# -*- coding: utf-8 -*-
"""FTP - File Transfer Protocol
==================================
.. module:: pcapkit.protocols.application.ftp.protocol
:mod:`pcapkit.protocols.application.ftp.protocol` holds
:class:`~pcapkit.protocols.application.ftp.protocol.FTP`, the extractor for the
control channel of the File Transfer Protocol (FTP) [*]_; the data channel is
:class:`~pcapkit.protocols.application.ftp_data.protocol.FTP_DATA`.
.. [*] https://en.wikipedia.org/wiki/File_Transfer_Protocol
.. versionchanged:: 1.5.0
A command or reply rebuilds byte for byte, its CRLF included, and keeps
the command's case and any octets that are not UTF-8 as captured. A line
with no argument and a multi-line reply, whole in one segment or not,
parse and rebuild as captured (:issue:`1238`, :issue:`1239`,
:issue:`1240`, :issue:`1241`, :issue:`1303`, :issue:`1333`).
.. versionremoved:: 1.5.0
``pcapkit.protocols.application.ftp.FTP_DATA``, with no alias. The class
moved to :mod:`pcapkit.protocols.application.ftp_data.protocol`, since TCP port 20
dispatches it and port 21 the control channel -- two protocols, so two
modules (:issue:`2285`).
"""
import re
from typing import TYPE_CHECKING
from pcapkit.const.ftp.command import Command as Enum_Command
from pcapkit.const.ftp.return_code import ReturnCode as Enum_ReturnCode
from pcapkit.corekit.enum import EnumLookup
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.protocols.application.application import Application
from pcapkit.protocols.application.ftp.data import FTP as Data_FTP
from pcapkit.protocols.application.ftp.data import Request as Data_Request
from pcapkit.protocols.application.ftp.data import Response as Data_Response
from pcapkit.protocols.application.ftp.schema import FTP as Schema_FTP
from pcapkit.utilities.chardet import detect
from pcapkit.utilities.compat import StrEnum, auto
from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall
if TYPE_CHECKING:
from typing import Any, Iterable, NoReturn, Optional
from typing_extensions import Literal
from pcapkit.const.reg.apptype.tcp import TCP as AppType_TCP
__all__ = ['FTP']
#: The TCP port registry :meth:`FTP.__index__` reads its port from, resolved
#: there rather than imported, as ``import pcapkit`` leaves it unloaded
#: (:issue:`1538`), so its first import holds
#: :data:`~pcapkit.utilities.importlib.import_lock` (:issue:`2771`).
_TCP: 'ModuleDescriptor[AppType_TCP]' = ModuleDescriptor('pcapkit.const.reg.apptype.tcp', 'TCP')
# FTP lines (RFC 959, sections 4 and 5): a command is ``<command> [<SP>
# <argument>] <CRLF>``, a reply ``<code> [<SP> <text>] <CRLF>``, and a multi-line
# reply opens ``<code>-<text> <CRLF>``. Only the one SP is syntax, so ``make``
# writes back the line as read. ``\Z``, as ``$`` also matches before a ``\n``.
FTP_REQUEST = re.compile(rb'^(?P<cmmd>[A-Z]{3,4})(?: (?P<args>.*))?\r\n\Z', re.I)
FTP_RESPONSE = re.compile(rb'^(?P<code>[0-9]{3})'
rb'(?:(?P<more>-)(?P<text>.*)|(?: (?P<args>.*))?)\r\n\Z', re.I)
[docs]
class Type(EnumLookup, StrEnum):
"""FTP packet type.
Built on :class:`~pcapkit.corekit.enum.EnumLookup`, the lookup contract
shared by every non-registry enumeration (:issue:`877`). The class defines
neither ``get`` nor ``_missing_`` of its own.
"""
#: Request packet.
REQUEST = auto()
#: Response packet.
RESPONSE = auto()
[docs]
class FTP(Application[Data_FTP, Schema_FTP],
data=Data_FTP, schema=Schema_FTP):
"""This class implements File Transfer Protocol."""
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'Literal["File Transfer Protocol"]':
"""Name of current protocol."""
return 'File Transfer Protocol'
@property
def length(self) -> 'NoReturn':
"""Header length of current protocol.
Raises:
UnsupportedCall: This protocol doesn't support :attr:`length`.
"""
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'length'")
##########################################################################
# Methods.
##########################################################################
[docs]
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_FTP': # pylint: disable=unused-argument
"""Read File Transfer Protocol (FTP).
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Raises:
ProtocolError: If the packet is malformed.
"""
if length is None:
length = len(self)
schema = self.__header__
data = schema.data
# NOTE: A multi-line reply (:rfc:`959#section-4`) may arrive in one segment; its
# first line is parsed, the rest kept in ``lines``, and anything else refused.
line, lines = data, () # type: bytes, tuple[bytes, ...]
if FTP_RESPONSE.match(data) is None and data.endswith(b'\r\n'):
first, _, rest = data.partition(b'\r\n')
if rest and (match := FTP_RESPONSE.match(first + b'\r\n')) is not None \
and match.group('more'):
rest_lines = tuple(rest[:-2].split(b'\r\n'))
if self._test_lines(match.group('code'), rest_lines):
line, lines = first + b'\r\n', rest_lines
if (match := FTP_REQUEST.match(line)) is not None:
raw_cmmd = match.group('cmmd')
args = match.group('args')
cmmd_val = Enum_Command.get(raw_cmmd.decode())
args_val, charset = self._decode_args(args)
ftp = Data_Request(
type=Type.REQUEST,
cmmd=cmmd_val,
args=args_val,
charset=charset,
raw_cmmd=raw_cmmd,
) # type: Data_FTP
elif (match := FTP_RESPONSE.match(line)) is not None:
code = int(match.group('code'))
more = bool(match.group('more'))
args = match.group('text') if more else match.group('args')
code_val = Enum_ReturnCode.get(code)
args_val, charset = self._decode_args(args)
ftp = Data_Response(
type=Type.RESPONSE,
code=code_val,
more=more,
args=args_val,
charset=charset,
lines=lines,
)
else:
raise ProtocolError('FTP: invalid packet format')
return ftp
[docs]
def make(self,
cmmd: 'Optional[Enum_Command | str | bytes | bytearray]' = None,
code: 'Optional[Enum_ReturnCode | int | str | bytes | bytearray]' = None,
args: 'Optional[str | bytes | bytearray]' = None,
more: 'bool' = False,
charset: 'Optional[str]' = None,
raw_cmmd: 'Optional[bytes]' = None,
lines: 'Iterable[bytes]' = (),
**kwargs: 'Any') -> 'Schema_FTP':
"""Make (construct) packet data.
The line is terminated with CRLF. Arguments of :obj:`None` write no
argument and no separator; otherwise a single SP separates them from
the command or code, except after the hyphen of a multi-line reply,
which the text follows immediately (:rfc:`959#section-4`).
Commands are case-insensitive (:rfc:`959#section-5`), so ``raw_cmmd``
is written in place of ``cmmd`` whenever the two differ only in case,
which keeps the case the command was read in.
Args:
cmmd: FTP command, a :class:`str` (an
:class:`~pcapkit.const.ftp.command.Command` member is one),
:class:`bytes` or :class:`bytearray`.
code: FTP status code, an :class:`int` in ``0``--``999`` (an
:class:`~pcapkit.const.ftp.return_code.ReturnCode` member is
one) or a numeric :class:`str`, :class:`bytes` or
:class:`bytearray` holding one.
args: Optional FTP command arguments and/or status messages, a
:class:`str`, :class:`bytes` or :class:`bytearray`.
more: More status messages to follow for response packets.
charset: Encoding for :obj:`str` arguments, **UTF-8** if not given.
raw_cmmd: FTP command as received, in its original case.
lines: Lines of a multi-line reply after the first, each written
verbatim and terminated with CRLF; the last, and only the last,
is the closing line, ``<code> SP <text>``.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Raises:
ProtocolError: If neither or both of ``cmmd`` and ``code`` are
given, if ``cmmd`` or ``args`` is not a :class:`str`,
:class:`bytes` or :class:`bytearray` (a :obj:`bool` included),
if ``code`` is not an :class:`int` in ``0``--``999`` (a
:obj:`bool` included) nor a numeric :class:`str`,
:class:`bytes` or :class:`bytearray` holding one, or if
``lines`` are given other than as the rest of a multi-line
reply.
"""
if cmmd is not None and code is None:
# NOTE: ``Command`` is a ``StrEnum``, so a member takes the ``str``
# branch; anything else is the wrong type (#2290).
if isinstance(cmmd, (bytes, bytearray)):
prefix = bytes(cmmd)
elif isinstance(cmmd, str):
prefix = cmmd.encode()
else:
raise ProtocolError(f'FTP: invalid command: must be a str, bytes or bytearray, not {cmmd!r}')
if raw_cmmd is not None and raw_cmmd.upper() == prefix.upper():
prefix = raw_cmmd
mf = b''
elif cmmd is None and code is not None:
prefix = self._make_code(code)
mf = b'-' if more else b''
else:
raise ProtocolError('FTP: invalid packet type')
lines = tuple(lines)
if lines and not (mf and self._test_lines(prefix, lines)):
raise ProtocolError('FTP: lines must close a multi-line reply')
# NOTE: bytes and a bytearray are written verbatim and only a ``str`` is encoded;
# anything else is the wrong type, as it is for ``cmmd`` above (#2410).
if args is None:
suffix = b''
elif isinstance(args, (bytes, bytearray)):
suffix = bytes(args)
elif isinstance(args, str):
suffix = args.encode(charset or 'utf-8')
else:
raise ProtocolError('FTP: invalid arguments: '
f'must be a str, bytes or bytearray, not {args!r}')
# the separating SP, absent with no arguments and after a hyphen
sep = b'' if args is None or mf else b' '
return Schema_FTP(
data=b'%s%s%s%s\r\n' % (prefix, mf, sep, suffix) + b''.join(b'%s\r\n' % line for line in lines),
)
##########################################################################
# Data models.
##########################################################################
[docs]
@classmethod
def __index__(cls) -> 'int': # type: ignore[override]
"""Numeral registry index of the protocol.
Returns:
Numeral registry index of the protocol in `IANA`_ service names and
port numbers: TCP port 21, the port of
:attr:`TCP.ftp <pcapkit.const.reg.apptype.tcp.TCP.ftp>` as an
:class:`int`, which
:attr:`TCP.__proto__ <pcapkit.protocols.transport.tcp.protocol.TCP.__proto__>`
dispatches here (:ref:`registry-protocol-index`).
.. _IANA: https://www.iana.org/assignments/service-names-port-numbers/service-names-port-numbers.xhtml
.. versionchanged:: 1.6.0
Returns TCP port 21, where it raised
:exc:`~pcapkit.utilities.exceptions.IntError` (:issue:`2627`).
"""
return int(_TCP.klass.ftp) # type: ignore[call-overload]
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_FTP') -> 'dict[str, Any]': # type: ignore[override]
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
return {
'cmmd': getattr(data, 'cmmd', None),
'code': getattr(data, 'code', None),
'args': getattr(data, 'args', None),
'more': getattr(data, 'more', False),
'charset': getattr(data, 'charset', None),
'raw_cmmd': getattr(data, 'raw_cmmd', None),
'lines': getattr(data, 'lines', ()),
}
@staticmethod
def _test_lines(code: 'bytes', lines: 'tuple[bytes, ...]') -> 'bool':
"""Whether ``lines`` are the rest of a multi-line reply with ``code``.
A multi-line reply ends at the first line that starts with its code
followed by SP (:rfc:`959#section-4`). The lines before that one are
free text, which may begin with digits -- the code with a hyphen
included -- but, as in any reply line, carry no LF.
Args:
code: Reply code of the first line.
lines: Lines after the first, CRLF excluded.
Returns:
Whether the last line, and only the last, is the closing line.
"""
closing = code + b' '
if not lines or not all(isinstance(line, bytes) and b'\n' not in line for line in lines):
return False
return lines[-1].startswith(closing) and not any(line.startswith(closing) for line in lines[:-1])
def _decode_args(self, args: 'Optional[bytes]') -> 'tuple[Optional[str], Optional[str]]':
"""Decode arguments, recording how to encode them back.
The arguments are decoded by :meth:`self.decode
<pcapkit.protocols.protocol.Protocol.decode>`. Where the result does
not encode back to ``args`` as UTF-8 -- which :meth:`make` uses by
default -- the detected charset is returned with it, or ``'latin-1'``
(decoding again) if that does not reproduce ``args`` either, so that the
original octets survive a rebuild.
Args:
args: Raw arguments, :obj:`None` if the line carries none.
Returns:
The decoded arguments and the charset to encode them with, the
latter :obj:`None` for UTF-8.
"""
if args is None:
return None, None
value = self.decode(args)
for charset in (None, detect(args)):
try:
if value.encode(charset or 'utf-8') == args:
return value, charset
except (LookupError, UnicodeError):
pass
return args.decode('latin-1'), 'latin-1'
@staticmethod
def _make_code(code: 'Enum_ReturnCode | int | str | bytes | bytearray') -> 'bytes':
"""Render a reply code as the three digits a reply line opens with.
The three digits hold ``0``--``999``, which is what the ``[0-9]{3}``
of ``FTP_RESPONSE`` reads back, so that is the range accepted rather
than the 1xx--5xx of :rfc:`959#section-4.2` alone -- 6xx is
:rfc:`2228#section-5`, and the read path accepts it.
Args:
code: Reply code, an :class:`int` (a
:class:`~pcapkit.const.ftp.return_code.ReturnCode` member is
one) or a numeric :class:`str`, :class:`bytes` or
:class:`bytearray` holding one.
Returns:
The code as three digits, a leading zero kept.
Raises:
ProtocolError: If ``code`` is of any other type, a :obj:`bool`
included, or does not fit in three digits.
"""
# NOTE: ``bool`` is refused first, or ``True`` would write 001; a bare ``int()``
# would raise ValueError or TypeError rather than ProtocolError (#2416).
value = None # type: Optional[int]
if isinstance(code, bool):
pass
elif isinstance(code, int):
value = code
elif isinstance(code, (str, bytes, bytearray)):
try:
value = int(code)
except ValueError:
pass
if value is None or not 0 <= value <= 999:
raise ProtocolError('FTP: invalid status code: must be an int in 0-999 or a '
f'numeric str, bytes or bytearray, not {code!r}')
return b'%03d' % value