Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 12 additions & 5 deletions pcapkit/const/ftp/return_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ def __new__(cls, value: 'int', description: 'Optional[str]' = None) -> 'Type[Ret
obj = int.__new__(cls, value)
obj._value_ = value

code = str(value)
code = '%03d' % value
obj.description = description
obj.kind = ResponseKind(int(code[0]))
obj.group = GroupingInformation(int(code[1]))
Expand Down Expand Up @@ -302,7 +302,7 @@ def _unregistered_member(cls, value: 'int', name: 'str') -> 'ReturnCode':

"""
obj = super()._unregistered_member(value, name)
code = str(value)
code = '%03d' % value
obj.description = None
obj.kind = ResponseKind(int(code[0]))
obj.group = GroupingInformation(int(code[1]))
Expand All @@ -312,11 +312,18 @@ def _unregistered_member(cls, value: 'int', name: 'str') -> 'ReturnCode':
def _missing_(cls, value: 'int') -> 'ReturnCode':
"""Lookup function used when value is not found.

A reply code is three digits (:rfc:`959#section-4.2`), so every value
from 0 to 999 can arrive on the wire: one within the registry's range
resolves to ``Unassigned``, any other to ``Unknown``.

Args:
value: Value to get enum item.

"""
if not (isinstance(value, int) and 100 <= value <= 659):
if not (isinstance(value, int) and 0 <= value <= 999):
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
#: Unassigned
return cls._unregistered_member(value, 'Unassigned')
if isinstance(value, int) and 100 <= value <= 659:
#: Unassigned
return cls._unregistered_member(value, 'Unassigned')
#: Unknown
return cls._unregistered_member(value, 'Unknown')
9 changes: 8 additions & 1 deletion pcapkit/const/http/status_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -273,12 +273,19 @@ def _unregistered_member(cls, value: 'int', name: 'str') -> 'StatusCode':
def _missing_(cls, value: 'int') -> 'StatusCode':
"""Lookup function used when value is not found.

A status code is three digits (:rfc:`9112#section-4`), so every value
from 0 to 999 can arrive on the wire: one within the registry's range
resolves through its unassigned ranges, any other to ``Unknown``.

Args:
value: Value to get enum item.

"""
if not (isinstance(value, int) and 100 <= value <= 599):
if not (isinstance(value, int) and 0 <= value <= 999):
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
if not (isinstance(value, int) and 100 <= value <= 599):
#: Unknown
return cls._unregistered_member(value, 'Unknown')
if 105 <= value <= 199:
#: Unassigned
return cls._unregistered_member(value, 'Unassigned')
Expand Down
2 changes: 1 addition & 1 deletion pcapkit/protocols/application/ftp.py
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ def make(self,
mf = b''
elif cmmd is None and code is not None:
code_val = int(code)
prefix = str(code_val).encode()
prefix = b'%03d' % code_val # reply codes are three digits, so keep a leading zero

mf = b'-' if more else b''
else:
Expand Down
3 changes: 2 additions & 1 deletion pcapkit/protocols/application/httpv1.py
Original file line number Diff line number Diff line change
Expand Up @@ -435,7 +435,8 @@ def _make_start_line(self, http_version: 'str | bytes', *,
if isinstance(msg, str):
msg = msg.encode()

header_line = b'HTTP/%s %s %s\r\n' % (version, str(status_code_val).encode(), msg)
# status-code is 3DIGIT (RFC 9112, section 4), so keep a leading zero
header_line = b'HTTP/%s %03d %s\r\n' % (version, status_code_val, msg)
else:
raise ProtocolError('HTTP packet must be either request or response.')
return header_line
Expand Down
17 changes: 12 additions & 5 deletions pcapkit/vendor/ftp/return_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ def __new__(cls, value: 'int', description: 'Optional[str]' = None) -> 'Type[{NA
obj = int.__new__(cls, value)
obj._value_ = value

code = str(value)
code = '%03d' % value
obj.description = description
obj.kind = ResponseKind(int(code[0]))
obj.group = GroupingInformation(int(code[1]))
Expand Down Expand Up @@ -158,7 +158,7 @@ def _unregistered_member(cls, value: 'int', name: 'str') -> '{NAME}':

"""
obj = super()._unregistered_member(value, name)
code = str(value)
code = '%03d' % value
obj.description = None
obj.kind = ResponseKind(int(code[0]))
obj.group = GroupingInformation(int(code[1]))
Expand All @@ -168,14 +168,21 @@ def _unregistered_member(cls, value: 'int', name: 'str') -> '{NAME}':
def _missing_(cls, value: 'int') -> '{NAME}':
"""Lookup function used when value is not found.

A reply code is three digits (:rfc:`959#section-4.2`), so every value
from 0 to 999 can arrive on the wire: one within the registry's range
resolves to ``Unassigned``, any other to ``Unknown``.

Args:
value: Value to get enum item.

"""
if not ({FLAG}):
if not (isinstance(value, int) and 0 <= value <= 999):
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
#: Unassigned
return cls._unregistered_member(value, 'Unassigned')
if {FLAG}:
#: Unassigned
return cls._unregistered_member(value, 'Unassigned')
#: Unknown
return cls._unregistered_member(value, 'Unknown')
''' # type: Callable[[str, str, str, str, str], str]


Expand Down
9 changes: 8 additions & 1 deletion pcapkit/vendor/http/status_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,12 +96,19 @@ def _unregistered_member(cls, value: 'int', name: 'str') -> '{NAME}':
def _missing_(cls, value: 'int') -> '{NAME}':
"""Lookup function used when value is not found.

A status code is three digits (:rfc:`9112#section-4`), so every value
from 0 to 999 can arrive on the wire: one within the registry's range
resolves through its unassigned ranges, any other to ``Unknown``.

Args:
value: Value to get enum item.

"""
if not ({FLAG}):
if not (isinstance(value, int) and 0 <= value <= 999):
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
if not ({FLAG}):
#: Unknown
return cls._unregistered_member(value, 'Unknown')
{MISS}
{'' if (test := ''.join(MISS.splitlines()[-1:])).startswith('return') or test[8:].startswith('return') else 'return super()._missing_(value)'}
'''.strip() # type: Callable[[str, str, str, str, str, str], str]
Expand Down
11 changes: 5 additions & 6 deletions tests/const/test_const_enum_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,13 +74,12 @@

#: Fully qualified names of the :class:`~aenum.IntEnum` classes under
#: :mod:`pcapkit.const` for which rejecting ``0`` is *correct*, because their
#: registries simply do not define ``0`` (three-digit response codes starting
#: at 100, or registries reserving ``0`` and starting real assignments at 1).
#: Confirmed against the IANA/RFC registries backing each one; see GitHub
#: issue #492.
#: registries simply do not define ``0`` (registries reserving ``0`` and
#: starting real assignments at 1). Confirmed against the IANA/RFC registries
#: backing each one; see GitHub issue #492. FTP ``ReturnCode`` and HTTP
#: ``StatusCode`` left the set with GitHub issue #1443: their wire fields are
#: three digits, so ``000`` resolves to an ``Unknown`` pseudo-member.
EXPECTED_TO_REJECT_ZERO = frozenset({
'pcapkit.const.ftp.return_code.ReturnCode',
'pcapkit.const.http.status_code.StatusCode',
'pcapkit.const.ftp.command.ConformanceRequirement',
'pcapkit.const.sctp.cause_code.CauseCode',
'pcapkit.const.sctp.parameter.Parameter',
Expand Down
126 changes: 126 additions & 0 deletions tests/protocols/application/test_ftp_http_unassigned_code_1443_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
# -*- coding: utf-8 -*-
"""FTP reply codes and HTTP status codes parse and rebuild for every three digits.

GitHub issue #1443: :class:`~pcapkit.const.ftp.return_code.ReturnCode` resolved
only 100 to 659 and :class:`~pcapkit.const.http.status_code.StatusCode` only
100 to 599, so a reply such as ``999 hi`` or a status line such as
``HTTP/1.1 999 X`` raised :exc:`~pcapkit.utilities.exceptions.EnumValueError`
and could not be parsed at all. Both wire fields are three digits, so every
value from 000 to 999 now resolves, to ``Unassigned`` inside the registry's
range and ``Unknown`` outside it, and the packet rebuilds byte for byte.
``make()`` also writes the code as three digits, so ``099`` keeps its zero.

Every case builds its own octets in memory and reads no capture. Classes are
imported inside each test, after
:func:`~tests._support.reimport_once_per_class`, so they match the live
:mod:`pcapkit` import.

"""

import unittest

from tests._support import reimport_once_per_class


class TestFTPReturnCodeEveryValue(unittest.TestCase):
"""Pin FTP replies across the whole three-digit range."""

def setUp(self) -> None:
reimport_once_per_class(self)

def test_every_three_digit_reply_rebuilds_byte_for_byte(self) -> None:
from pcapkit.protocols.application.ftp import FTP

for code in range(1000):
raw = b'%03d hi\r\n' % code
with self.subTest(code=raw):
parsed = FTP(raw, len(raw))
self.assertEqual(int(parsed.info.code), code)
self.assertEqual(FTP.from_data(parsed.info).data, raw)

def test_multi_line_unassigned_reply_rebuilds_byte_for_byte(self) -> None:
from pcapkit.protocols.application.ftp import FTP

for raw in (b'777-first\r\nmiddle\r\n777 last\r\n', b'099-a\r\n099 b\r\n'):
with self.subTest(raw=raw):
parsed = FTP(raw, len(raw))
self.assertEqual(FTP.from_data(parsed.info).data, raw)

def test_unassigned_and_unknown_pseudo_members(self) -> None:
from pcapkit.const.ftp.return_code import ReturnCode

cases = ((199, 'Unassigned', 1, 9), (659, 'Unassigned', 6, 5),
(777, 'Unknown', 7, 7), (99, 'Unknown', 0, 9), (0, 'Unknown', 0, 0))
for value, name, kind, group in cases:
with self.subTest(value=value):
member = ReturnCode.get(value)
self.assertEqual(int(member), value)
self.assertEqual(member.name, name)
self.assertEqual(int(member.kind), kind)
self.assertEqual(int(member.group), group)
self.assertNotIn(name, ReturnCode.__members__)
self.assertEqual(str(member), '[%d] None' % value)

def test_value_outside_three_digits_still_raises(self) -> None:
from pcapkit.const.ftp.return_code import ReturnCode

for value in (-1, 1000):
with self.subTest(value=value):
with self.assertRaisesRegex(ValueError, r'%d is not a valid ReturnCode' % value):
ReturnCode.get(value)

def test_make_keeps_the_leading_zero(self) -> None:
from pcapkit.protocols.application.ftp import FTP

proto = FTP(code=99, args='x')
self.assertEqual(proto.data, b'099 x\r\n')
self.assertEqual(FTP(proto.data, len(proto.data)).data, b'099 x\r\n')


class TestHTTPStatusCodeEveryValue(unittest.TestCase):
"""Pin HTTP/1.1 status lines across the whole three-digit range."""

def setUp(self) -> None:
reimport_once_per_class(self)

def test_every_three_digit_status_rebuilds_byte_for_byte(self) -> None:
from pcapkit.protocols.application.httpv1 import HTTP

for code in range(1000):
raw = b'HTTP/1.1 %03d X\r\n\r\n' % code
with self.subTest(code=code):
parsed = HTTP(raw, len(raw))
self.assertEqual(int(parsed.info.receipt.status), code)
self.assertEqual(HTTP.from_data(parsed.info).data, raw)

def test_unassigned_and_unknown_pseudo_members(self) -> None:
from pcapkit.const.http.status_code import StatusCode

for value, name in ((999, 'Unknown'), (600, 'Unknown'), (99, 'Unknown'),
(0, 'Unknown'), (599, 'Unassigned'), (105, 'Unassigned')):
with self.subTest(value=value):
member = StatusCode.get(value)
self.assertEqual(int(member), value)
self.assertEqual(member.name, name)
self.assertEqual(member.message, name)
self.assertNotIn(name, StatusCode.__members__)

def test_value_outside_three_digits_still_raises(self) -> None:
from pcapkit.const.http.status_code import StatusCode

for value in (-1, 1000):
with self.subTest(value=value):
with self.assertRaisesRegex(ValueError, r'%d is not a valid StatusCode' % value):
StatusCode.get(value)

def test_make_keeps_the_leading_zero(self) -> None:
from pcapkit.protocols.application.httpv1 import HTTP

proto = HTTP(status=99, message='X', http_version='1.1')
self.assertEqual(proto.data, b'HTTP/1.1 099 X\r\n\r\n')
reparsed = HTTP(proto.data, len(proto.data))
self.assertEqual(int(reparsed.info.receipt.status), 99)


if __name__ == '__main__':
unittest.main()
Loading