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
37 changes: 25 additions & 12 deletions pcapkit/const/reg/apptype/apptype.py
Original file line number Diff line number Diff line change
Expand Up @@ -2390,7 +2390,10 @@ def get(cls, key: 'int', *,
ValueError: If called on a class that holds no members, i.e. on
:class:`AppType` itself, with a ``proto`` naming no registry to
delegate to. Also for a ``key`` that is not a port number, since
this registry resolves ports and not service names.
this registry resolves ports and not service names -- including
one outside ``0..65535``, whose rejection by :meth:`_missing_` this
method propagates rather than minting over, so that ``get`` is
never more permissive than ``AppType(...)``.

:meta private:
"""
Expand All @@ -2409,11 +2412,15 @@ def get(cls, key: 'int', *,
# answered with.
return matched[0]

try:
ret = owner._missing_(key)
if ret is None:
raise ValueError
except ValueError:
# NOTE: :meth:`_missing_` answers :obj:`None` for a port it holds no row
# for, which is what minting is for, and *raises* for a value that is not a
# port at all. Catching that rejection was GitHub issue #758's defect: it
# minted ``PORT_999999_tcp`` and ``PORT_-1_tcp``, the latter a name no
# attribute access can reach, and left ``get`` more permissive than
# ``AppType(...)``, which has always raised here. The rejection now
# propagates, so both entry points answer an out-of-range port identically.
ret = owner._missing_(key)
if ret is None:
ret = extend_enum(owner, 'PORT_%d_%s' % (key, owner.__transport__.name),
key, 'unknown', owner.__transport__)
return ret
Expand Down Expand Up @@ -2460,10 +2467,16 @@ def _missing_(cls, value: 'int') -> 'Optional[AppType]':
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
# NOTE: extending this class would give it a member, and aenum then
# refuses to subclass it -- permanently, for every registry not yet
# imported. The spans below are IANA's unassigned ranges, which belong to
# whichever registry was asked, never to this one.
# imported. The spans below belong to whichever registry was asked, never
# to this one.
if cls.__registry__ is None:
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
# NOTE: most spans are IANA's unassigned and reserved markers, which name
# no transport protocol and so answer every registry. A span that does name
# one tests ``cls.__transport__`` and answers that registry alone --
# GitHub issue #760, where source order decided instead and a UDP lookup in
# 6000-6063 came back carrying ``tcp``. A registry a named span excludes
# falls through to a mint, which is what IANA assigning it nothing means.
if 225 <= value <= 241:
#: [N/A] Reserved [:rfc:`1060`]
return extend_enum(cls, 'reserved_%d' % value, value, 'reserved', TransportProtocol.get('undefined'))
Expand Down Expand Up @@ -2887,10 +2900,10 @@ def _missing_(cls, value: 'int') -> 'Optional[AppType]':
if 5995 <= value <= 5998:
#: [N/A] Unassigned
return extend_enum(cls, 'unassigned_%d' % value, value, 'unassigned', TransportProtocol.get('undefined'))
if 6000 <= value <= 6063:
if 6000 <= value <= 6063 and cls.__transport__ is TransportProtocol.get('tcp'):
#: [TCP] X Window System
return extend_enum(cls, 'x11_%d' % value, value, 'x11', TransportProtocol.get('tcp'))
if 6000 <= value <= 6063:
if 6000 <= value <= 6063 and cls.__transport__ is TransportProtocol.get('udp'):
#: [UDP] X Window System
return extend_enum(cls, 'x11_%d' % value, value, 'x11', TransportProtocol.get('udp'))
if 6078 <= value <= 6079:
Expand Down Expand Up @@ -3037,10 +3050,10 @@ def _missing_(cls, value: 'int') -> 'Optional[AppType]':
if 6658 <= value <= 6664:
#: [N/A] Unassigned
return extend_enum(cls, 'unassigned_%d' % value, value, 'unassigned', TransportProtocol.get('undefined'))
if 6665 <= value <= 6669:
if 6665 <= value <= 6669 and cls.__transport__ is TransportProtocol.get('tcp'):
#: [TCP] IRCU
return extend_enum(cls, 'ircu_%d' % value, value, 'ircu', TransportProtocol.get('tcp'))
if 6665 <= value <= 6669:
if 6665 <= value <= 6669 and cls.__transport__ is TransportProtocol.get('udp'):
#: [UDP] Reserved
return extend_enum(cls, 'reserved_%d' % value, value, 'reserved', TransportProtocol.get('udp'))
if 6674 <= value <= 6677:
Expand Down
50 changes: 40 additions & 10 deletions pcapkit/vendor/reg/apptype/apptype.py
Original file line number Diff line number Diff line change
Expand Up @@ -345,7 +345,10 @@ def get(cls, key: 'int', *,
ValueError: If called on a class that holds no members, i.e. on
:class:`{NAME}` itself, with a ``proto`` naming no registry to
delegate to. Also for a ``key`` that is not a port number, since
this registry resolves ports and not service names.
this registry resolves ports and not service names -- including
one outside ``0..65535``, whose rejection by :meth:`_missing_` this
method propagates rather than minting over, so that ``get`` is
never more permissive than ``{NAME}(...)``.

:meta private:
"""
Expand All @@ -364,11 +367,15 @@ def get(cls, key: 'int', *,
# answered with.
return matched[0]

try:
ret = owner._missing_(key)
if ret is None:
raise ValueError
except ValueError:
# NOTE: :meth:`_missing_` answers :obj:`None` for a port it holds no row
# for, which is what minting is for, and *raises* for a value that is not a
# port at all. Catching that rejection was GitHub issue #758's defect: it
# minted ``PORT_999999_tcp`` and ``PORT_-1_tcp``, the latter a name no
# attribute access can reach, and left ``get`` more permissive than
# ``{NAME}(...)``, which has always raised here. The rejection now
# propagates, so both entry points answer an out-of-range port identically.
ret = owner._missing_(key)
if ret is None:
ret = extend_enum(owner, 'PORT_%d_%s' % (key, owner.__transport__.name),
key, 'unknown', owner.__transport__)
return ret
Expand Down Expand Up @@ -415,10 +422,16 @@ def _missing_(cls, value: 'int') -> 'Optional[{NAME}]':
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
# NOTE: extending this class would give it a member, and aenum then
# refuses to subclass it -- permanently, for every registry not yet
# imported. The spans below are IANA's unassigned ranges, which belong to
# whichever registry was asked, never to this one.
# imported. The spans below belong to whichever registry was asked, never
# to this one.
if cls.__registry__ is None:
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
# NOTE: most spans are IANA's unassigned and reserved markers, which name
# no transport protocol and so answer every registry. A span that does name
# one tests ``cls.__transport__`` and answers that registry alone --
# GitHub issue #760, where source order decided instead and a UDP lookup in
# 6000-6063 came back carrying ``tcp``. A registry a named span excludes
# falls through to a mint, which is what IANA assigning it nothing means.
{MISS}
{'' if ''.join(MISS.splitlines()[-1:]).startswith('return') else 'return super()._missing_(value)'}
'''.strip() # type: Callable[[str, str, str, str, str, str], str]
Expand Down Expand Up @@ -684,10 +697,27 @@ def records(self, data: 'list[str]') -> 'tuple[OrderedDict[str, Record], list[st
except ValueError:
start, stop = port.split('-')

miss.append(f'if {start} <= value <= {stop}:')
# NOTE: a span IANA assigns to a transport protocol is claimed by
# that registry alone, through a test on ``cls.__transport__``.
# Source order decided instead until GitHub issue #760: the only
# two spans registered on more than one transport -- 6000-6063 and
# 6665-6669 -- rendered the same condition twice, leaving the
# second copy unreachable, so *every* registry answered with the
# first row's. That is not merely a duplicate to collapse: the
# 6665-6669 rows are two different services, ``ircu`` on TCP and
# IANA's ``reserved`` marker on UDP, so one merged branch carrying
# ``tcp | udp`` would have to discard one of them.
#
# A span naming no transport protocol is an unassigned or reserved
# marker belonging to whichever registry was asked, so it carries
# no test and keeps answering all four.
flag = self.flag([proto])
claim = '' if proto == 'undefined' else f' and cls.__transport__ is {flag}'

miss.append(f'if {start} <= value <= {stop}{claim}:')
miss.append(f' #: {cmmt}')
miss.append(f" return extend_enum(cls, '{self.safe_name(svc)}_%d' % value, "
f"value, {svc!r}, {self.flag([proto])})")
f"value, {svc!r}, {flag})")

return line, miss

Expand Down
102 changes: 102 additions & 0 deletions tests/const/test_const_apptype_split_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,108 @@ def test_a_real_port_with_no_transport_no_longer_resolves(self) -> None:
# It is still documented, just not resolvable.
self.assertIn('``reserved``', AppType.__doc__ or '')

def test_get_rejects_an_out_of_range_port_as_the_constructor_does(self) -> None:
"""GitHub issue #758: ``get`` swallowed ``_missing_``'s own rejection.

``test_every_registry_rejects_a_negative_value`` above asserts ``cls(-1)``
and ``cls(65536)`` and **never** ``.get``, which is how CI stayed green
while ``AppType.get(-1, proto='tcp')`` minted ``PORT_-1_tcp`` -- a key no
attribute access can reach, since it is not an identifier. ``_missing_``
did raise; ``get``'s ``except ValueError`` caught that very rejection and
minted regardless, leaving ``get`` more permissive than ``cls(...)``. So
the assertion has to be on ``.get``, and on the registry not having grown:
the defect was visible as ``len(TCP)`` going 6147 to 6148.
"""
from pcapkit.const.reg.apptype import DCCP, SCTP, TCP, UDP, AppType, TransportProtocol

for cls in (TCP, UDP, SCTP, DCCP):
for port in (-1, 65536, 999999):
with self.subTest(registry=cls.__name__, port=port):
before = len(cls)
with self.assertRaises(ValueError):
cls.get(port)
with self.assertRaises(ValueError):
cls.get_all(port)
with self.assertRaises(ValueError):
AppType.get(port, proto=cls.__transport__)
self.assertEqual(len(cls), before)
self.assertNotIn('PORT_%d_%s' % (port, cls.__transport__.name),
cls.__members__)

# Paid for by narrowing nothing: a valid but unassigned port is what the
# mint is *for*, and it still answers with one.
minted = AppType.get(59000, proto=TransportProtocol.tcp)
self.addCleanup(self._purge_member, TCP, 'PORT_59000_tcp', 59000)
self.assertEqual(minted.svc, 'unknown')
self.assertEqual(minted.port, 59000)

def test_a_span_on_two_transports_answers_each_with_its_own_row(self) -> None:
"""GitHub issue #760: source order decided, so the TCP row answered all four.

IANA registers exactly two port *spans* on more than one transport
protocol, and the crawler rendered one ``_missing_`` branch per registry
row -- two branches with an identical condition, the second unreachable.
Every registry therefore answered with the first: ``AppType.get(6010,
proto='udp')`` returned a **UDP** member whose ``proto`` read ``tcp``.

6665-6669 is why the duplicate is not simply collapsed into one branch
carrying ``tcp | udp``: its two rows are different services -- ``ircu`` on
TCP, IANA's ``reserved`` marker on UDP -- so a merged branch would have to
discard one of them, and #732's ``TransportProtocol`` retype would then
want a *named* combination for it. Each branch tests ``cls.__transport__``
instead, which also stays right for a span naming one transport only.
"""
from pcapkit.const.reg.apptype import SCTP, TCP, UDP, AppType

for cls, port, svc in ((TCP, 6010, 'x11'), (UDP, 6010, 'x11'),
(TCP, 6666, 'ircu'), (UDP, 6666, 'reserved')):
with self.subTest(registry=cls.__name__, port=port):
resolved = AppType.get(port, proto=cls.__transport__)
self.addCleanup(self._purge_member, cls, '%s_%d' % (svc, port), port)
self.assertIs(type(resolved), cls)
self.assertEqual(resolved.svc, svc)
self.assertIs(resolved.proto, cls.__transport__)

# A span IANA assigns to TCP and UDP assigns nothing to SCTP, so the third
# registry mints rather than inheriting the TCP row -- which it did, as
# ``<SCTP.x11: 6010 [tcp]>``.
resolved = AppType.get(6010, proto='sctp')
self.addCleanup(self._purge_member, SCTP, 'PORT_6010_sctp', 6010)
self.assertEqual(resolved.svc, 'unknown')
self.assertIs(resolved.proto, SCTP.__transport__)

def test_every_transport_named_span_is_claimed_by_one_registry(self) -> None:
"""The structural half of #760, which is what survives the next crawl.

:meth:`~pcapkit.const.reg.apptype.apptype.AppType._missing_` is generated,
so the defect returns the moment
:mod:`pcapkit.vendor.reg.apptype.apptype` stops emitting the test -- and
it returns silently, because a shadowed branch is unreachable rather than
wrong. This asserts the invariant over every branch instead of over the two
spans that happen to collide today: a branch minting a member for a named
transport protocol is claimed by that registry, and one minting for
``undefined`` claims nothing, since IANA's unassigned and reserved markers
belong to whichever registry was asked.
"""
import inspect
import re

from pcapkit.const.reg.apptype import AppType

source = inspect.getsource(AppType._missing_.__func__) # type: ignore[attr-defined]
branches = re.findall(r'\n if (.+?):\n #:.*?\n '
r'return extend_enum\(.+?TransportProtocol\.get\((.+?)\)\)',
source)
self.assertEqual(len(branches), 766)

for condition, proto in branches:
with self.subTest(condition=condition):
claim = 'cls.__transport__ is TransportProtocol.get(%s)' % proto
if proto == "'undefined'":
self.assertNotIn('cls.__transport__', condition)
else:
self.assertIn(claim, condition)

@staticmethod
def _purge_member(cls: type, name: str, port: int) -> None:
"""Undo an :func:`~aenum.extend_enum` so the registry is left as found.
Expand Down
Loading