fix(ftp,httpv1): FTP and HTTP/1 lines do not rebuild byte for byte

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<cmmd>[A-Z]{3,4})( +(?P<args>.*))?\r\n$', re.I)
-FTP_RESPONSE = re.compile(rb'^(?P<code>[0-9]{3})(?P<more>\-)?( +(?P<args>.*))?\r\n$', re.I)
+# regex for FTP, per :rfc:`959#section-4` (replies) and :rfc:`959#section-5`
+# (commands): a command is ``<command> [<SP> <argument>] <CRLF>``, a reply line is
+# ``<code> [<SP> <text>] <CRLF>``, and the first line of a multi-line reply is
+# ``<code>-<text> <CRLF>`` 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<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)
 
 
 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
+        <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'
+
 
 class FTP_DATA(Raw):
     """This class implements FTP data channel transmission."""

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
+        <pcapkit.protocols.protocol.Protocol.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')

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

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