diff --git a/pcapkit/protocols/application/ftp.py b/pcapkit/protocols/application/ftp.py index 4f0deac3d0..09319f9e33 100644 --- a/pcapkit/protocols/application/ftp.py +++ b/pcapkit/protocols/application/ftp.py @@ -26,6 +26,7 @@ from pcapkit.protocols.data.application.ftp import Response as Data_Response from pcapkit.protocols.misc.raw import Raw from pcapkit.protocols.schema.application.ftp 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 @@ -36,9 +37,16 @@ __all__ = ['FTP', 'FTP_DATA'] -# regex for FTP -FTP_REQUEST = re.compile(rb'^(?P[A-Z]{3,4})( +(?P.*))?\r\n$', re.I) -FTP_RESPONSE = re.compile(rb'^(?P[0-9]{3})(?P\-)?( +(?P.*))?\r\n$', re.I) +# regex for FTP, per :rfc:`959#section-4` (replies) and :rfc:`959#section-5` +# (commands): a command is `` [ ] ``, a reply line is +# `` [ ] ``, and the first line of a multi-line reply is +# ``- `` with the text following the hyphen immediately. Only +# the one separating SP is syntax; anything after it is the argument verbatim, so +# that ``make`` writes back exactly the line that was read. ``\Z`` rather than +# ``$``, which would also match before a trailing ``\n`` left outside the match. +FTP_REQUEST = re.compile(rb'^(?P[A-Z]{3,4})(?: (?P.*))?\r\n\Z', re.I) +FTP_RESPONSE = re.compile(rb'^(?P[0-9]{3})' + rb'(?:(?P-)(?P.*)|(?: (?P.*))?)\r\n\Z', re.I) class Type(EnumLookup, StrEnum): @@ -107,26 +115,28 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_FTP': args = match.group('args') cmmd_val = Enum_Command.get(cmmd) - args_val = self.decode(args) + args_val, charset = self._decode_args(args) ftp = Data_Request( type=Type.REQUEST, cmmd=cmmd_val, args=args_val, + charset=charset, ) # type: Data_FTP elif (match := FTP_RESPONSE.match(data)) is not None: code = int(match.group('code')) more = bool(match.group('more')) - args = match.group('args') + args = match.group('text') if more else match.group('args') code_val = Enum_ReturnCode.get(code) - args_val = self.decode(args) + args_val, charset = self._decode_args(args) ftp = Data_Response( type=Type.RESPONSE, code=code_val, more=more, args=args_val, + charset=charset, ) else: raise ProtocolError('FTP: invalid packet format') @@ -137,14 +147,21 @@ def make(self, code: 'Optional[Enum_ReturnCode | int | str | bytes]' = None, args: 'Optional[str | bytes]' = None, more: 'bool' = False, + charset: 'Optional[str]' = None, **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`). + Args: cmmd: FTP command. code: FTP status code. args: Optional FTP command arguments and/or status messages. more: More status messages to follow for response packets. + charset: Encoding for :obj:`str` arguments, **UTF-8** if not given. **kwargs: Arbitrary keyword arguments. Returns: @@ -173,10 +190,12 @@ def make(self, elif isinstance(args, bytes): suffix = args else: - suffix = args.encode() + suffix = args.encode(charset or 'utf-8') + # 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' % (prefix, mf, suffix), + data=b'%s%s%s%s\r\n' % (prefix, mf, sep, suffix), ) ########################################################################## @@ -199,8 +218,38 @@ def _make_data(cls, data: 'Data_FTP') -> 'dict[str, Any]': # type: ignore[overr 'code': getattr(data, 'code', None), 'args': getattr(data, 'args', None), 'more': getattr(data, 'more', False), + 'charset': getattr(data, 'charset', None), } + 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 + `. 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' + class FTP_DATA(Raw): """This class implements FTP data channel transmission.""" diff --git a/pcapkit/protocols/application/httpv1.py b/pcapkit/protocols/application/httpv1.py index 99e21f0093..e85b5cf120 100644 --- a/pcapkit/protocols/application/httpv1.py +++ b/pcapkit/protocols/application/httpv1.py @@ -40,6 +40,7 @@ from pcapkit.protocols.data.application.httpv1 import RequestHeader as Data_RequestHeader from pcapkit.protocols.data.application.httpv1 import ResponseHeader as Data_ResponseHeader from pcapkit.protocols.schema.application.httpv1 import HTTP as Schema_HTTP +from pcapkit.utilities.chardet import detect from pcapkit.utilities.compat import StrEnum, auto from pcapkit.utilities.exceptions import ProtocolError @@ -259,6 +260,7 @@ def make(self, # type: ignore[override] message: 'Optional[str | bytes]' = None, headers: 'Optional[OrderedMultiDict[str, str]]' = None, body: 'bytes' = b'', + charset: 'Optional[str]' = None, **kwargs: 'Any') -> 'Schema_HTTP': """Make (construct) packet data. @@ -273,6 +275,8 @@ def make(self, # type: ignore[override] message: HTTP status message. headers: HTTP headers. body: HTTP body. + charset: Encoding for a :obj:`str` request URI or status message, + **UTF-8** if not given. **kwargs: Arbitrary keyword arguments. Returns: @@ -292,7 +296,7 @@ def make(self, # type: ignore[override] meth = method.encode() else: meth = method.value.encode() - uri_val = uri.encode() if isinstance(uri, str) else uri + uri_val = uri.encode(charset or 'utf-8') if isinstance(uri, str) else uri header_line = b'%s %s HTTP/%s\r\n' % (meth, uri_val, version) elif method is None and status is not None: @@ -303,7 +307,7 @@ def make(self, # type: ignore[override] if message is None: msg = getattr(status, 'message', None) or getattr(status_code, 'message', b'') or b'' else: - msg = message.encode() if isinstance(message, str) else message + msg = message.encode(charset or 'utf-8') if isinstance(message, str) else message if isinstance(msg, str): msg = msg.encode() @@ -357,8 +361,36 @@ def _make_data(cls, data: 'Data_HTTP') -> 'dict[str, Any]': # type: ignore[over 'message': getattr(data.receipt, 'message', None), 'headers': data.header, 'body': b'' if data.body is None else data.body, + 'charset': getattr(data.receipt, 'charset', None), } + def _decode_text(self, text: 'bytes') -> 'tuple[str, Optional[str]]': + """Decode start-line text, recording how to encode it back. + + The text is decoded by :meth:`self.decode + `. Where the result does + not encode back to ``text`` 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 ``text`` either, so that the + original octets survive a rebuild. + + Args: + text: Raw request URI or status message. + + Returns: + The decoded text and the charset to encode it with, the latter + :obj:`None` for UTF-8. + + """ + value = self.decode(text) + for charset in (None, detect(text)): + try: + if value.encode(charset or 'utf-8') == text: + return value, charset + except (LookupError, UnicodeError): + pass + return text.decode('latin-1'), 'latin-1' + def _read_http_header(self, header: 'bytes') -> 'tuple[Data_Header, OrderedMultiDict[str, str]]': """Read HTTP/1.* header. @@ -448,18 +480,22 @@ def _read_http_header(self, header: 'bytes') -> 'tuple[Data_Header, OrderedMulti match3 = re.match(_RE_VERSION, para1) match4 = re.match(_RE_STATUS, para2) if match1 and match2: + uri, charset = self._decode_text(para2) header_line = Data_RequestHeader( type=Type.REQUEST, method=Enum_Method.get(self.decode(match1.group('method'))), - uri=self.decode(para2), + uri=uri, version=self.decode(match2.group('version')), + charset=charset, ) elif match3 and match4: + message, charset = self._decode_text(para3) header_line = Data_ResponseHeader( type=Type.RESPONSE, version=self.decode(match3.group('version')), status=Enum_StatusCode.get(int(para2)), - message=self.decode(para3), + message=message, + charset=charset, ) else: raise ProtocolError('HTTP: invalid format') diff --git a/pcapkit/protocols/data/application/ftp.py b/pcapkit/protocols/data/application/ftp.py index 20a5391f49..0e94014144 100644 --- a/pcapkit/protocols/data/application/ftp.py +++ b/pcapkit/protocols/data/application/ftp.py @@ -8,6 +8,8 @@ from pcapkit.protocols.data.protocol import Protocol if TYPE_CHECKING: + from typing import Optional + from typing_extensions import Literal from pcapkit.const.ftp.command import Command @@ -35,11 +37,13 @@ class Request(FTP): type: 'Literal[FTP_Type.REQUEST]' #: Command. cmmd: 'Command' - #: Arguments. - args: 'str' + #: Arguments, :obj:`None` if the command carries none. + args: 'Optional[str]' + #: Charset to encode :attr:`args` with, :obj:`None` for UTF-8. + charset: 'Optional[str]' if TYPE_CHECKING: - def __init__(self, type: 'Literal[FTP_Type.REQUEST]', cmmd: 'Command', args: 'str') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'Literal[FTP_Type.REQUEST]', cmmd: 'Command', args: 'Optional[str]', charset: 'Optional[str]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -50,10 +54,12 @@ class Response(FTP): type: 'Literal[FTP_Type.RESPONSE]' #: Return code. code: 'ReturnCode' - #: Arguments. - args: 'str' + #: Arguments, :obj:`None` if the reply carries no text. + args: 'Optional[str]' #: More data flag. more: 'bool' + #: Charset to encode :attr:`args` with, :obj:`None` for UTF-8. + charset: 'Optional[str]' if TYPE_CHECKING: - def __init__(self, type: 'Literal[FTP_Type.RESPONSE]', code: 'ReturnCode', args: 'str', more: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'Literal[FTP_Type.RESPONSE]', code: 'ReturnCode', args: 'Optional[str]', more: 'bool', charset: 'Optional[str]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin diff --git a/pcapkit/protocols/data/application/httpv1.py b/pcapkit/protocols/data/application/httpv1.py index 3a2fc2ca04..d1c8c19a52 100644 --- a/pcapkit/protocols/data/application/httpv1.py +++ b/pcapkit/protocols/data/application/httpv1.py @@ -8,7 +8,7 @@ from pcapkit.protocols.data.protocol import Protocol if TYPE_CHECKING: - from typing import Any + from typing import Any, Optional from pcapkit.const.http.method import Method as Enum_Method from pcapkit.const.http.status_code import StatusCode as Enum_StatusCode @@ -57,9 +57,11 @@ class RequestHeader(Header): uri: 'str' #: HTTP request version. version: 'str' + #: Charset to encode :attr:`uri` with, :obj:`None` for UTF-8. + charset: 'Optional[str]' if TYPE_CHECKING: - def __init__(self, type: 'HTTP_Type', method: 'Enum_Method', uri: 'str', version: 'str') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'HTTP_Type', method: 'Enum_Method', uri: 'str', version: 'str', charset: 'Optional[str]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -74,6 +76,8 @@ class ResponseHeader(Header): status: 'Enum_StatusCode' #: HTTP response status message. message: 'str' + #: Charset to encode :attr:`message` with, :obj:`None` for UTF-8. + charset: 'Optional[str]' if TYPE_CHECKING: - def __init__(self, type: 'HTTP_Type', version: 'str', status: 'Enum_StatusCode', message: 'str') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'HTTP_Type', version: 'str', status: 'Enum_StatusCode', message: 'str', charset: 'Optional[str]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin diff --git a/tests/protocols/application/test_ftp_http1_roundtrip_unit.py b/tests/protocols/application/test_ftp_http1_roundtrip_unit.py new file mode 100644 index 0000000000..5a539fef94 --- /dev/null +++ b/tests/protocols/application/test_ftp_http1_roundtrip_unit.py @@ -0,0 +1,145 @@ +# -*- coding: utf-8 -*- +"""FTP and HTTP/1.* lines rebuild byte for byte. C.f. #1238, #1239, #1240, #1241. + +- #1238: ``FTP.make`` wrote no CRLF, which the parse regexes require, so no FTP + line rebuilt at all. +- #1239: a line with no arguments parsed to ``args=None`` and then crashed in + ``decode``; ``make`` wrote a separating SP regardless. +- #1240: the text of a multi-line reply follows the hyphen immediately + (:rfc:`959#section-4`), but the parse required a SP there and ``make`` wrote + one. +- #1241: text decoded with a detected charset was encoded back as UTF-8, so + non-UTF-8 octets in FTP arguments and HTTP/1.* start lines changed on rebuild. + The charset is now recorded on the data whenever UTF-8 would not reproduce the + octets. + +""" + +from __future__ import annotations + +import importlib.util +import unittest + +from tests._support import reimport_once_per_class + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: FTP lines: name, octets. +FTP_LINES = ( + ('request', b'USER anonymous\r\n'), # 1238 + ('request without arguments', b'QUIT\r\n'), # 1239 + ('request with empty arguments', b'QUIT \r\n'), + ('request with leading space', b'USER anonymous\r\n'), + ('reply', b'230 Logged in\r\n'), + ('reply without text', b'220\r\n'), # 1239 + ('multi-line reply', b'220-Welcome\r\n'), # 1240 + ('multi-line reply with space', b'220- Welcome\r\n'), + ('multi-line reply without text', b'220-\r\n'), + ('latin-1 request', b'USER caf\xe9\r\n'), # 1241 + ('latin-1 reply', b'550 caf\xe9 cr\xe8me\r\n'), +) + +#: HTTP/1.* messages: name, octets. +HTTP1_MESSAGES = ( + ('latin-1 request URI', b'GET /caf\xe9 HTTP/1.1\r\n\r\n'), # 1241 + ('latin-1 reason phrase', b'HTTP/1.1 200 caf\xe9\r\n\r\n'), # 1241 + ('utf-8 request URI', b'GET /caf\xc3\xa9 HTTP/1.1\r\nHost: a\r\n\r\n'), +) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class FTPRoundTripTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_parse_then_rebuild_is_byte_exact(self) -> None: + from pcapkit.protocols.application.ftp import FTP + + for name, raw in FTP_LINES: + with self.subTest(name=name): + self.assertEqual(FTP.from_data(FTP(raw, len(raw)).info).data, raw) + + def test_make_terminates_with_crlf(self) -> None: + from pcapkit.protocols.application.ftp import FTP + + self.assertEqual(bytes(FTP(cmmd='USER', args='x')), b'USER x\r\n') + self.assertEqual(object.__new__(FTP).make(cmmd='USER', args='anonymous').data, + b'USER anonymous\r\n') + + def test_no_arguments(self) -> None: + from pcapkit.const.ftp.command import Command + from pcapkit.protocols.application.ftp import FTP + + request = FTP(b'QUIT\r\n', 6).info + self.assertIs(request.cmmd, Command.QUIT) + self.assertIsNone(request.args) + self.assertIsNone(FTP(b'220\r\n', 5).info.args) + self.assertEqual(object.__new__(FTP).make(cmmd='QUIT').data, b'QUIT\r\n') + + def test_multi_line_reply(self) -> None: + from pcapkit.protocols.application.ftp import FTP + + reply = FTP(b'220-Welcome\r\n', 13).info + self.assertTrue(reply.more) + self.assertEqual(reply.args, 'Welcome') + self.assertEqual(object.__new__(FTP).make(code=220, args='Welcome', more=True).data, + b'220-Welcome\r\n') + + def test_non_utf8_arguments_keep_their_charset(self) -> None: + from pcapkit.protocols.application.ftp import FTP + + raw = b'USER caf\xe9\r\n' + request = FTP(raw, len(raw)).info + self.assertEqual(request.args, 'café') + self.assertIsNotNone(request.charset) + self.assertEqual( + object.__new__(FTP).make(cmmd=request.cmmd, args=request.args, + charset=request.charset).data, + b'USER caf\xe9\r\n', + ) + # UTF-8 needs no charset recorded, and stays the default. + raw = b'USER caf\xc3\xa9\r\n' + self.assertIsNone(FTP(raw, len(raw)).info.charset) + self.assertEqual(object.__new__(FTP).make(cmmd='USER', args='café').data, + b'USER caf\xc3\xa9\r\n') + + def test_trailing_newline_is_not_dropped(self) -> None: + from pcapkit.protocols.application.ftp import FTP + from pcapkit.utilities.exceptions import ProtocolError + + ftp = object.__new__(FTP) + ftp.__cached__ = {} + for raw in (b'USER x\r\n\n', b'220 x\r\n\n'): + with self.subTest(raw=raw): + ftp.__header__ = type('Header', (), {'data': raw})() + with self.assertRaises(ProtocolError): + ftp.read(length=len(raw)) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class HTTP1RoundTripTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_parse_then_rebuild_is_byte_exact(self) -> None: + from pcapkit.protocols.application.httpv1 import HTTP + + for name, raw in HTTP1_MESSAGES: + with self.subTest(name=name): + self.assertEqual(HTTP.from_data(HTTP(raw, len(raw)).info).data, raw) + + def test_non_utf8_start_line_keeps_its_charset(self) -> None: + from pcapkit.protocols.application.httpv1 import HTTP + + raw = b'GET /caf\xe9 HTTP/1.1\r\n\r\n' + receipt = HTTP(raw, len(raw)).info.receipt + self.assertEqual(receipt.uri, '/café') + self.assertIsNotNone(receipt.charset) + + raw = b'GET /caf\xc3\xa9 HTTP/1.1\r\n\r\n' + self.assertIsNone(HTTP(raw, len(raw)).info.receipt.charset) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/application/test_ftp_unit.py b/tests/protocols/application/test_ftp_unit.py index bff5e4a1eb..54f575cf0f 100644 --- a/tests/protocols/application/test_ftp_unit.py +++ b/tests/protocols/application/test_ftp_unit.py @@ -52,12 +52,14 @@ def test_ftp_read_parses_request_response_and_rejects_invalid(self) -> None: self.assertEqual(request.cmmd, Command.USER) self.assertEqual(request.args, 'anonymous') + # GitHub issue #1240: the text of a multi-line reply follows the + # hyphen immediately (:rfc:`959#section-4`), so a SP there is text. ftp.__header__ = SimpleNamespace(data=b'220- Ready\r\n') response = ftp.read(length=12) self.assertEqual(response.type, Type.RESPONSE) self.assertEqual(response.code, ReturnCode.CODE_220) self.assertTrue(response.more) - self.assertEqual(response.args, 'Ready') + self.assertEqual(response.args, ' Ready') ftp.__header__ = SimpleNamespace(data=b'!!\r\n') with self.assertRaises(ProtocolError): @@ -176,11 +178,13 @@ def test_ftp_make_builds_request_and_response_packets(self) -> None: response = ftp.make(code=ReturnCode.CODE_220, args='Ready', more=True) response_bytes = ftp.make(code=b'230', args=b'Logged in') - self.assertEqual(request.data, b'USER anonymous') - self.assertEqual(request_bytes.data, b'NOOP ') - self.assertEqual(request_value.data, b'PWD ') - self.assertEqual(response.data, b'220- Ready') - self.assertEqual(response_bytes.data, b'230 Logged in') + # GitHub issues #1238, #1239 and #1240: every line ends in CRLF, no + # separator is written without arguments, and none after a hyphen. + self.assertEqual(request.data, b'USER anonymous\r\n') + self.assertEqual(request_bytes.data, b'NOOP\r\n') + self.assertEqual(request_value.data, b'PWD\r\n') + self.assertEqual(response.data, b'220-Ready\r\n') + self.assertEqual(response_bytes.data, b'230 Logged in\r\n') def test_ftp_make_rejects_ambiguous_packet_type(self) -> None: from pcapkit.protocols.application.ftp import FTP @@ -198,16 +202,18 @@ def test_ftp_make_data_preserves_request_and_response_fields(self) -> None: from pcapkit.const.ftp.return_code import ReturnCode from pcapkit.protocols.application.ftp import FTP - request = DummyData(cmmd=Command.USER, code=None, args='anonymous', more=False) - response = DummyData(cmmd=None, code=ReturnCode.CODE_220, args='Ready', more=True) + request = DummyData(cmmd=Command.USER, code=None, args='anonymous', more=False, charset=None) + response = DummyData(cmmd=None, code=ReturnCode.CODE_220, args='Ready', more=True, + charset='latin-1') self.assertEqual( FTP._make_data(request), - {'cmmd': Command.USER, 'code': None, 'args': 'anonymous', 'more': False}, + {'cmmd': Command.USER, 'code': None, 'args': 'anonymous', 'more': False, 'charset': None}, ) self.assertEqual( FTP._make_data(response), - {'cmmd': None, 'code': ReturnCode.CODE_220, 'args': 'Ready', 'more': True}, + {'cmmd': None, 'code': ReturnCode.CODE_220, 'args': 'Ready', 'more': True, + 'charset': 'latin-1'}, ) def test_ftp_data_channel_name_is_stable(self) -> None: