From 764caa1f69cc0e8731a4c029899df4413699c877 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 16 Sep 2026 13:40:12 -0400 Subject: [PATCH 1/6] foundation: recover a failed next layer through the protocol, not the schema `beholder` read the fallback payload from `self.__header__.get_payload()`. That is what `ProtocolBase._get_payload()` wraps, so the two agree for every protocol whose payload is a schema field -- and disagree for the two that override it. SCTP carries user data inside a DATA chunk and PCAP-NG inside a block, so neither header schema has a `payload` field at all, and reaching for the schema raises `ProtocolUnbound("unknown field: 'payload'")` *from the recovery path*. A next-layer parse failure that should have degraded to `Raw` became a crash that aborted the frame. It was unreachable until now only because `SCTP.__proto__` had no entries: with nothing registered on a payload protocol identifier, no next-layer parse could fail. Registering anything on one exposes it immediately. - `beholder` now calls `self._get_payload()`, the wrapper both overrides. - It also forwards `alias=proto`, which the success path in `_import_next_layer` already passes. Without it, a payload that failed to parse came back as a bare `Raw` while an *unregistered* number came back named -- so registering a protocol made the output less informative than leaving the number alone. A plain integer has no `name` and still renders as `Raw`, so this only adds a name where the registry key is an enumeration. The four `beholder` tests are rewritten onto shared stand-ins that carry a `_get_payload`, as a real protocol does, plus two new ones: the alias forwarding and an overridden `_get_payload` whose schema accessor raises, which is the SCTP shape. Unit tier green, and mypy reports the same five pre-existing errors in `decorators.py` as it does at HEAD -- no new ones. --- pcapkit/utilities/decorators.py | 29 +++++- tests/utilities/test_decorators.py | 154 ++++++++++++++++++----------- 2 files changed, 121 insertions(+), 62 deletions(-) diff --git a/pcapkit/utilities/decorators.py b/pcapkit/utilities/decorators.py index f170fb732b..7cb8bb68c8 100644 --- a/pcapkit/utilities/decorators.py +++ b/pcapkit/utilities/decorators.py @@ -111,6 +111,10 @@ def beholder(func: 'Callable[Concatenate[Protocol, int, Optional[int], P], R_beh def behold(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_beholder': # extract self object & args self = cast('R_beholder', args[0]) + try: + proto = args[1] + except IndexError: + proto = None try: length = cast('int', args[2]) except IndexError: @@ -134,8 +138,29 @@ def behold(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_beholder': logger.error('The following error occurred while parsing the packet:') traceback.print_exc() - file_ = self.__header__.get_payload() - next_ = protocol(file_, length, error=str(exc)) + # NOTE: ``self._get_payload()`` rather than + # ``self.__header__.get_payload()``, which it wraps. The two agree + # for every protocol whose payload is a schema field, and differ for + # the two that override it: SCTP carries user data inside a DATA + # chunk and PCAP-NG inside a block, so neither header schema has a + # ``payload`` field at all. Going through the schema there raises + # ProtocolUnbound('unknown field: payload') *from the recovery path*, + # turning a next-layer parse failure that should have degraded to + # Raw into a crash. Unreachable until something was registered on an + # SCTP payload protocol identifier, which NGAP now is. + file_ = self._get_payload() + + # NOTE: ``alias=proto`` matches what the success path passes, so a + # payload that failed to parse is still named after the protocol + # number it arrived with -- ``SCTP:PayloadProtocolIdentifier_3GPP_NG + # _Application_Protocol`` rather than a bare ``SCTP:Raw``. Without + # it, registering a protocol on a number made the output *less* + # informative than leaving the number unregistered, since an + # unregistered number reaches Raw through the success path and keeps + # its name. A plain integer has no ``name`` and still renders as + # ``Raw``, c.f. ``Raw.__post_init__``, so this only adds a name where + # the registry key is an enumeration. + next_ = protocol(file_, length, error=str(exc), alias=proto) return cast('R_beholder', next_) return behold diff --git a/tests/utilities/test_decorators.py b/tests/utilities/test_decorators.py index f600615379..83801ef7a0 100644 --- a/tests/utilities/test_decorators.py +++ b/tests/utilities/test_decorators.py @@ -167,30 +167,63 @@ def unpack(cls, data, length=None, packet=None): self.assertIsInstance(returned, dict) self.assertTrue(returned['prepped']) - def test_beholder_wraps_struct_eof_with_no_payload(self) -> None: - exceptions = self.exceptions - + ########################################################################## + # beholder. + # + # The stand-in protocols below carry a ``_get_payload`` that delegates to + # ``self.__header__.get_payload()``, because that is what + # ``ProtocolBase._get_payload`` is -- a wrapper over the schema's accessor. + # ``beholder`` goes through the wrapper rather than the schema directly, so + # that the two protocols which override it (SCTP, PCAP-NG) recover from a + # next-layer failure at all; see + # ``test_beholder_uses_the_protocol_payload_accessor_not_the_schema``. + ########################################################################## + + @staticmethod + def _payload_stand_ins(): + """``Raw`` and ``NoPayload`` stand-ins recording what they were handed.""" class NoPayload: - def __init__(self, file_, length, error=None) -> None: + def __init__(self, file_, length, error=None, alias=None) -> None: self.file = file_ self.length = length self.error = error + self.alias = alias class Raw(NoPayload): pass install_fake_payload_protocols(Raw, NoPayload) + return Raw, NoPayload + @staticmethod + def _demo_protocol_class(decorator, payload: bytes, raises, drop_length=False): + """A protocol whose next-layer decode always raises ``raises``.""" class Header: def get_payload(self) -> bytes: - return b'payload-bytes' + return payload class DemoProtocol: __header__ = Header() - @self.decorators.beholder - def decode(self, proto, length=None): - raise exceptions.StructError('unexpected eof', eof=True) + def _get_payload(self) -> bytes: + return self.__header__.get_payload() + + if drop_length: + @decorator + def decode(self, proto): + raise raises + else: + @decorator + def decode(self, proto, length=None): + raise raises + + return DemoProtocol + + def test_beholder_wraps_struct_eof_with_no_payload(self) -> None: + Raw, NoPayload = self._payload_stand_ins() + DemoProtocol = self._demo_protocol_class( + self.decorators.beholder, b'payload-bytes', + self.exceptions.StructError('unexpected eof', eof=True)) result = DemoProtocol().decode(1, 10) @@ -200,27 +233,9 @@ def decode(self, proto, length=None): self.assertEqual(result.error, 'unexpected eof') def test_beholder_wraps_other_errors_with_raw(self) -> None: - class NoPayload: - def __init__(self, file_, length, error=None) -> None: - self.file = file_ - self.length = length - self.error = error - - class Raw(NoPayload): - pass - - install_fake_payload_protocols(Raw, NoPayload) - - class Header: - def get_payload(self) -> bytes: - return b'raw-bytes' - - class DemoProtocol: - __header__ = Header() - - @self.decorators.beholder - def decode(self, proto, length=None): - raise ValueError('broken parser') + Raw, _ = self._payload_stand_ins() + DemoProtocol = self._demo_protocol_class( + self.decorators.beholder, b'raw-bytes', ValueError('broken parser')) result = DemoProtocol().decode(1, 3) @@ -229,29 +244,65 @@ def decode(self, proto, length=None): self.assertEqual(result.length, 3) self.assertEqual(result.error, 'broken parser') - def test_beholder_verbose_mode_prints_traceback(self) -> None: - class NoPayload: - def __init__(self, file_, length, error=None) -> None: - self.file = file_ - self.length = length - self.error = error + def test_beholder_forwards_the_protocol_number_as_an_alias(self) -> None: + """A failed payload keeps the number it arrived with. - class Raw(NoPayload): - pass + The success path in ``_import_next_layer`` passes ``alias=proto``, so a + payload that reached :class:`Raw` because nothing was registered on its + number is still labelled with that number. Omitting it from the failure + path made *registering* a protocol produce less informative output than + leaving the number unregistered. - install_fake_payload_protocols(Raw, NoPayload) + """ + Raw, _ = self._payload_stand_ins() + DemoProtocol = self._demo_protocol_class( + self.decorators.beholder, b'raw-bytes', ValueError('broken parser')) + + self.assertEqual(DemoProtocol().decode(60, 3).alias, 60) + # No ``proto`` argument at all -- the decorator must not raise. + DemoProtocol = self._demo_protocol_class( + self.decorators.beholder, b'raw-bytes', ValueError('broken parser'), + drop_length=True) + self.assertEqual(DemoProtocol().decode(66).alias, 66) + + def test_beholder_uses_the_protocol_payload_accessor_not_the_schema(self) -> None: + """An overridden ``_get_payload`` is what the recovery path reads. + + SCTP carries user data inside a DATA chunk and PCAP-NG inside a block, + so neither header schema has a ``payload`` field: reaching for the schema + raises ``ProtocolUnbound('unknown field: payload')`` *from the recovery + path*, turning a next-layer failure that should have degraded to + :class:`Raw` into a crash. It was unreachable until something was + registered on an SCTP payload protocol identifier, which NGAP now is. + + """ + exceptions = self.exceptions + Raw, _ = self._payload_stand_ins() class Header: def get_payload(self) -> bytes: - return b'raw-bytes' + raise exceptions.ProtocolUnbound("unknown field: 'payload'") class DemoProtocol: __header__ = Header() + def _get_payload(self) -> bytes: + return b'chunk-user-data' + @self.decorators.beholder def decode(self, proto, length=None): raise ValueError('broken parser') + result = DemoProtocol().decode(60, 15) + + self.assertIsInstance(result, Raw) + self.assertEqual(result.file, b'chunk-user-data') + + def test_beholder_verbose_mode_prints_traceback(self) -> None: + Raw, _ = self._payload_stand_ins() + DemoProtocol = self._demo_protocol_class( + self.decorators.beholder, b'raw-bytes', ValueError('broken parser')) + from unittest import mock with mock.patch.object(self.decorators, 'VERBOSE', True): @@ -262,27 +313,10 @@ def decode(self, proto, length=None): print_exc.assert_called_once() def test_beholder_defaults_length_when_argument_is_missing(self) -> None: - class NoPayload: - def __init__(self, file_, length, error=None) -> None: - self.file = file_ - self.length = length - self.error = error - - class Raw(NoPayload): - pass - - install_fake_payload_protocols(Raw, NoPayload) - - class Header: - def get_payload(self) -> bytes: - return b'raw-bytes' - - class DemoProtocol: - __header__ = Header() - - @self.decorators.beholder - def decode(self, proto): - raise ValueError('broken parser') + Raw, _ = self._payload_stand_ins() + DemoProtocol = self._demo_protocol_class( + self.decorators.beholder, b'raw-bytes', ValueError('broken parser'), + drop_length=True) result = DemoProtocol().decode(1) From bd8eac4b2148519440996bde3e74ee82dc5c2629 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 16 Sep 2026 13:40:36 -0400 Subject: [PATCH 2/6] protocols: implement NGAP over SCTP, decoding aligned PER through pycrate (#251) NGAP (3GPP TS 38.413) is the 5G RAN-to-AMF control plane. It has no header of its own: an SCTP DATA chunk whose payload protocol identifier names NGAP carries exactly one aligned-PER `NGAP-PDU` and nothing else, so nothing about it is visible until an ASN.1 decoder has run. - `pcapkit.protocols.application.ngap`, with the matching schema and data models, and exports added everywhere `FTP` is listed. - Registered on `SCTP.__proto__` as a *default*, for PPID 60 (`NG_Application_Protocol`) and 66 (`NGAP_over_DTLS_over_SCTP`), following how `TCP.__proto__` declares 21 and 80 rather than calling `register_sctp` at import time from somewhere. - `pycrate` is an optional extra, `pip install pypcapkit[NGAP]`, imported inside the parse path and never at module import, and deliberately excluded from `all` -- it lands ~238 MB of `pycrate_asn1dir` (87 spec modules) to obtain the one 4.9 MB `NGAP.py`, and it is LGPL-2.1+ where this package is BSD-3-Clause. Absent, it fails the way `pcapkit.protocols.internet.esp` fails without `cryptography`: a `ProtocolError` that `beholder` degrades to `Raw`. - Decoding is generic rather than per-procedure: the value tree is mapped by ASN.1 *shape*, so all 81 elementary procedures and 438 protocol IEs work without 519 hand-written cases and a new 3GPP release needs no code change. The PDU kind, procedure code, criticality, message type name and IE list are surfaced as first-class fields. - `ProcedureCode`, `Criticality` and `ProtocolIE` live in `ngap.py`, not `pcapkit.const`: 3GPP publishes them in the ASN.1 of TS 38.413 rather than in a crawlable registry, so there is no vendor module either. The SCTP PPIDs are IANA's and are used from `pcapkit.const.sctp` as they stand. `reset_val()` is never called on the parse path, and there is a comment plus a test saying so: it walks all 318 submodules of the compiled specification and costs ~102 ms against the decode's ~0.15 ms, and `from_aper()` overwrites the value anyway. The module-level PDU object is stateful and `get_val()` hands back its own containers, so a lock is held across the decode *and* the conversion. Five SCTP tests used PPID 60 as an arbitrary placeholder with junk payloads and now describe the registered default instead; their `SCTP.__proto__.clear()` teardowns are replaced with a snapshot restore, since clearing the registry now destroys the defaults for every later test in the process. Unit tier: 670 passed with pycrate, 613 passed / 65 skipped without it, no failures either way. Round trip verified byte-exact on a real 58-octet `NGSetupRequest`. --- pcapkit/__init__.py | 2 +- pcapkit/all.py | 2 +- pcapkit/protocols/__init__.py | 1 + pcapkit/protocols/application/__init__.py | 6 + pcapkit/protocols/application/ngap.py | 1124 +++++++++++++++++ pcapkit/protocols/data/__init__.py | 4 + .../protocols/data/application/__init__.py | 11 + pcapkit/protocols/data/application/ngap.py | 123 ++ pcapkit/protocols/schema/__init__.py | 1 + .../protocols/schema/application/__init__.py | 6 + pcapkit/protocols/schema/application/ngap.py | 29 + pcapkit/protocols/transport/sctp.py | 15 +- pyproject.toml | 23 + tests/protocols/application/test_ngap_unit.py | 487 +++++++ tests/protocols/transport/test_sctp_unit.py | 97 +- 15 files changed, 1903 insertions(+), 28 deletions(-) create mode 100644 pcapkit/protocols/application/ngap.py create mode 100644 pcapkit/protocols/data/application/ngap.py create mode 100644 pcapkit/protocols/schema/application/ngap.py create mode 100644 tests/protocols/application/test_ngap_unit.py diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index 7c6e706b6d..d9de56a0aa 100644 --- a/pcapkit/__init__.py +++ b/pcapkit/__init__.py @@ -117,7 +117,7 @@ 'TCP', 'UDP', 'SCTP', # Transport Layer 'FTP', 'FTP_DATA', # Application Layer - 'HTTP', + 'HTTP', 'NGAP', ] #: version number diff --git a/pcapkit/all.py b/pcapkit/all.py index c4c179f1ea..c1ffe62c37 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -115,7 +115,7 @@ # IPv6 Extension Header 'TCP', 'UDP', 'SCTP', # Transport Layer 'FTP', 'FTP_DATA', # Application Layer - 'HTTP', + 'HTTP', 'NGAP', 'Schema', 'schema', # Protocol Schema 'Data', 'data', # Protocol Data diff --git a/pcapkit/protocols/__init__.py b/pcapkit/protocols/__init__.py index 3e61edeb55..b132b5513a 100644 --- a/pcapkit/protocols/__init__.py +++ b/pcapkit/protocols/__init__.py @@ -66,6 +66,7 @@ # Application Layer 'FTP', 'FTP_DATA', 'HTTP', 'HTTPv1', 'HTTPv2', + 'NGAP', ] #: dict[str, Type[Protocol]]: Protocol registry. diff --git a/pcapkit/protocols/application/__init__.py b/pcapkit/protocols/application/__init__.py index 19c5b1ff01..064ef28f95 100644 --- a/pcapkit/protocols/application/__init__.py +++ b/pcapkit/protocols/application/__init__.py @@ -12,6 +12,10 @@ # TODO: Implements BGP, DHCP, DHCPv6, DNS, IMAP, LDAP, MQTT, # NNTP, NTP, ONC:RPC, POP, RIP, RTP, SIP, SMTP, SNMP, # SSH, TELNET, TLS/SSL, XMPP. +# +# NB: NGAP below is the one protocol here whose decoder is an optional +# dependency, c.f. pcapkit.protocols.application.ngap. Importing the module +# is free -- ``pycrate`` is imported inside the parse path, not here. # Base Class for Internet Layer from pcapkit.protocols.application.application import Application @@ -20,6 +24,7 @@ from pcapkit.protocols.application.ftp import FTP, FTP_DATA from pcapkit.protocols.application.httpv1 import HTTP as HTTPv1 from pcapkit.protocols.application.httpv2 import HTTP as HTTPv2 +from pcapkit.protocols.application.ngap import NGAP # Deprecated / Base Classes from pcapkit.protocols.application.http import HTTP @@ -31,4 +36,5 @@ 'APPTYPE', 'FTP', 'FTP_DATA', 'HTTP', 'HTTPv1', 'HTTPv2', + 'NGAP', ] diff --git a/pcapkit/protocols/application/ngap.py b/pcapkit/protocols/application/ngap.py new file mode 100644 index 0000000000..589946fd04 --- /dev/null +++ b/pcapkit/protocols/application/ngap.py @@ -0,0 +1,1124 @@ +# -*- coding: utf-8 -*- +# pylint: disable=line-too-long +"""NGAP - NG application protocol +==================================== + +.. module:: pcapkit.protocols.application.ngap + +:mod:`pcapkit.protocols.application.ngap` contains +:class:`~pcapkit.protocols.application.ngap.NGAP` only, +which implements extractor for the NG Application Protocol +(NGAP) [*]_, as specified in 3GPP TS 38.413. + +NGAP is the control plane between a 5G RAN node (gNB or ng-eNB) and an AMF. +It runs over SCTP and is named by the DATA chunk's *payload protocol +identifier* rather than by a port, so :class:`NGAP` is registered on +:attr:`SCTP.__proto__ ` under +PPID 60 (``NG_Application_Protocol``) and PPID 66 +(``NGAP_over_DTLS_over_SCTP``), c.f. +:func:`pcapkit.foundation.registry.protocols.register_sctp`. + +An SCTP DATA chunk that names NGAP carries exactly one ``NGAP-PDU``, encoded +in **aligned PER** (ALIGNED PACKED ENCODING RULES, :abbr:`APER`). There is no +header to read and no framing to resolve: the whole payload is the encoding, +and none of its structure is visible until an ASN.1 decoder has run over it. + +Decoding therefore needs the optional |pycrate|_ dependency +(``pip install pypcapkit[NGAP]``). :mod:`pcapkit` imports and works without +it; an NGAP payload simply degrades to the opaque payload path, exactly as an +unregistered PPID would, because +:meth:`SCTP._import_next_layer ` +is wrapped in :func:`~pcapkit.utilities.decorators.beholder` and falls back to +:class:`~pcapkit.protocols.misc.raw.Raw`. + +Why |pycrate|_ rather than a PER codec of our own +------------------------------------------------- + +Two things make it the cheaper answer. |pycrate|_ **ships NGAP already +compiled**, at ``pycrate_asn1dir/NGAP.py``, so the 3GPP ASN.1 source does not +have to be vendored here and tracked across releases; and it is **pure +Python**, with no compiled extension to build on any platform. Decoding costs +0.15 ms per PDU, the same order as :mod:`pcapkit`'s own per-packet cost, so +the generic strategy below is not paying for the convenience. + +Generic conversion, not 51 hand-written procedures +-------------------------------------------------- + +The decoded value tree is mapped into :class:`~pcapkit.corekit.infoclass.Info` +objects **structurally**, by ASN.1 shape rather than by procedure: + +=============================== ========================================================== +ASN.1 / |pycrate|_ shape :mod:`pcapkit` model +=============================== ========================================================== +``SEQUENCE`` / ``SET`` (a dict) :class:`~pcapkit.protocols.data.application.ngap.Sequence` +``SEQUENCE OF`` (a list) :obj:`list` +``CHOICE`` / open type :class:`~pcapkit.protocols.data.application.ngap.Choice` +``BIT STRING`` :class:`~pcapkit.protocols.data.application.ngap.BitString` +``INTEGER``, ``OCTET STRING``, kept as :obj:`int`, :obj:`bytes`, :obj:`str` +``ENUMERATED``, ``BOOLEAN`` +=============================== ========================================================== + +That is a deliberate trade. Every one of the 81 elementary procedures and 438 +protocol IEs works on the day it is decoded, and a new 3GPP release needs no +code change here; what is given up is per-IE typing, so an IE's value is +reported in the specification's own shape rather than as a +:mod:`pcapkit`-specific model. The fields worth reading at a glance -- the PDU +kind, procedure code, criticality, message type name and the IE list -- are +surfaced as first-class fields on +:class:`~pcapkit.protocols.data.application.ngap.NGAP` regardless. + +Known limitations +----------------- + +* **PPID 66 payloads are not decoded.** ``NGAP_over_DTLS_over_SCTP`` wraps the + ``NGAP-PDU`` in a DTLS record, and :mod:`pcapkit` implements no DTLS, so the + bytes reaching :meth:`NGAP.read` are not an APER encoding. The PPID is + registered so that it is *named* rather than anonymous; the payload itself + degrades to :class:`~pcapkit.protocols.misc.raw.Raw`. +* **The specification version is |pycrate|_'s, not this package's.** The IE and + procedure enumerations below were generated from ``NGAP_Constants`` of + |pycrate|_ 0.8.1 (Release-18-era: 81 procedure codes, 438 protocol IE IDs, + highest 443). A |pycrate|_ that carries a newer NGAP will decode IEs that + :class:`ProcedureCode` and :class:`ProtocolIE` do not name; both extend + themselves at lookup time rather than failing, so such a value is reported + as ``Unassigned_``. +* **NGAP over a fragmented SCTP association is not reassembled.** A DATA chunk + is decoded on its own, so an ``NGAP-PDU`` split across chunks by SCTP + fragmentation fails to decode rather than being reassembled first. +* **Private IEs (``PrivateMessage``) carry no schema.** Their contents are + vendor defined, so the generic conversion reports whatever ASN.1 shape the + encoding declares and cannot name the fields. + +.. |pycrate| replace:: ``pycrate`` +.. _pycrate: https://github.com/pycrate-org/pycrate + +.. [*] https://en.wikipedia.org/wiki/NG_Application_Protocol + +""" +import threading +from typing import TYPE_CHECKING + +from aenum import IntEnum, extend_enum + +from pcapkit.corekit.infoclass import Info +from pcapkit.protocols.application.application import Application +from pcapkit.protocols.data.application.ngap import IE as Data_IE +from pcapkit.protocols.data.application.ngap import NGAP as Data_NGAP +from pcapkit.protocols.data.application.ngap import BitString as Data_BitString +from pcapkit.protocols.data.application.ngap import Choice as Data_Choice +from pcapkit.protocols.data.application.ngap import Sequence as Data_Sequence +from pcapkit.protocols.schema.application.ngap import NGAP as Schema_NGAP +from pcapkit.utilities.compat import StrEnum +from pcapkit.utilities.exceptions import ProtocolError + +if TYPE_CHECKING: + from typing import Any, NoReturn, Optional + + from typing_extensions import Literal + +__all__ = ['NGAP', 'PDUKind', 'Criticality', 'ProcedureCode', 'ProtocolIE'] + +#: Cached ``NGAP-PDU`` object, c.f. :func:`load_pycrate`. :data:`NotImplemented` +#: means the import has not been attempted yet, and :data:`None` that it was +#: attempted and |pycrate|_ is not installed -- three states, so a capture full +#: of NGAP packets does not pay for a failing import on every frame. +_PYCRATE = NotImplemented # type: Any + +#: Guards the module-level ``NGAP-PDU`` object returned by :func:`load_pycrate`. +#: That object is *stateful*: :meth:`from_aper` stores the decoded value on it +#: and :meth:`get_val` hands back the decoder's own containers rather than +#: copies, so two decodes running concurrently through it would each see the +#: other's tree. The lock is held across the conversion, not merely across the +#: decode, for that second reason. +_PDU_LOCK = threading.Lock() + + +def load_pycrate() -> 'Optional[Any]': + """Load the optional |pycrate|_ ``NGAP-PDU`` object. + + Returns: + ``pycrate_asn1dir.NGAP.NGAP_PDU_Descriptions.NGAP_PDU``, the ``CHOICE`` + over ``initiatingMessage`` / ``successfulOutcome`` / + ``unsuccessfulOutcome`` that is the entry point of the compiled + specification, or :data:`None` when |pycrate|_ is not installed. + + Notes: + The import is attempted at most once and the outcome is cached. It is + not free even when it succeeds -- ``pycrate_asn1dir.NGAP`` is a 4.9 MB + module -- which is the other reason it happens here rather than at + module import: neither ``import pcapkit`` nor the documentation build + should pay for it. + + """ + global _PYCRATE # pylint: disable=global-statement + + if _PYCRATE is NotImplemented: + try: + from pycrate_asn1dir.NGAP import \ + NGAP_PDU_Descriptions # pylint: disable=import-outside-toplevel + except ImportError: + _PYCRATE = None + else: + _PYCRATE = NGAP_PDU_Descriptions.NGAP_PDU + return _PYCRATE + + +############################################################################## +# Enumerations. +# +# These are 3GPP TS 38.413 specification values, not an IANA registry with a +# crawlable page, so they live here rather than in pcapkit.const and have no +# vendor crawler. The SCTP payload protocol identifiers *are* IANA's, and are +# used from pcapkit.const.sctp.payload_protocol_identifier rather than +# duplicated here. +# +# The two large enumerations below were generated from pycrate 0.8.1's +# ``NGAP_Constants`` at authoring time and pasted in. Nothing at import time +# needs pycrate to define them. +############################################################################## + + +class PDUKind(StrEnum): + """Which alternative of the ``NGAP-PDU`` ``CHOICE`` a PDU is. + + The values are spelled as the ASN.1 identifiers, so that a name decoded by + |pycrate|_ resolves by value. + + """ + + #: A procedure's request, or a class 2 procedure's only message. + INITIATING_MESSAGE = 'initiatingMessage' + #: A class 1 procedure's successful response. + SUCCESSFUL_OUTCOME = 'successfulOutcome' + #: A class 1 procedure's unsuccessful response. + UNSUCCESSFUL_OUTCOME = 'unsuccessfulOutcome' + + +class Criticality(IntEnum): + """[Criticality] What a receiver must do with an IE it does not understand. + + Members are named for the ASN.1 identifiers rather than upper-cased, so + that :meth:`Criticality.get` resolves a name decoded by |pycrate|_ through + the standard member map. The values are the ``ENUMERATED`` indices, which + is what goes on the wire. + + """ + + #: Reject the whole message. + reject = 0 + #: Ignore the IE and carry on. + ignore = 1 + #: Ignore the IE, carry on, and report it. + notify = 2 + + @staticmethod + def get(key: 'int | str | Criticality', default: 'int' = -1) -> 'Criticality': + """Backport support for original codes. + + Args: + key: Key to get enum item. + default: Default value if not found. + + :meta private: + """ + if isinstance(key, Criticality): + return key + if isinstance(key, int): + return Criticality(key) + return Criticality[key] # type: ignore[misc] + + @classmethod + def _missing_(cls, value: 'int') -> 'NoReturn': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + Raises: + ValueError: Always. ``Criticality`` is an ``ENUMERATED`` with no + extension marker, so a fourth value cannot be encoded and a + lookup for one is a bug rather than a newer specification. + + """ + raise ValueError('%r is not a valid %s' % (value, cls.__name__)) + + +class ProcedureCode(IntEnum): + """[ProcedureCode] NGAP elementary procedure codes, 3GPP TS 38.413.""" + + AMFConfigurationUpdate = 0 + AMFStatusIndication = 1 + CellTrafficTrace = 2 + DeactivateTrace = 3 + DownlinkNASTransport = 4 + DownlinkNonUEAssociatedNRPPaTransport = 5 + DownlinkRANConfigurationTransfer = 6 + DownlinkRANStatusTransfer = 7 + DownlinkUEAssociatedNRPPaTransport = 8 + ErrorIndication = 9 + HandoverCancel = 10 + HandoverNotification = 11 + HandoverPreparation = 12 + HandoverResourceAllocation = 13 + InitialContextSetup = 14 + InitialUEMessage = 15 + LocationReportingControl = 16 + LocationReportingFailureIndication = 17 + LocationReport = 18 + NASNonDeliveryIndication = 19 + NGReset = 20 + NGSetup = 21 + OverloadStart = 22 + OverloadStop = 23 + Paging = 24 + PathSwitchRequest = 25 + PDUSessionResourceModify = 26 + PDUSessionResourceModifyIndication = 27 + PDUSessionResourceRelease = 28 + PDUSessionResourceSetup = 29 + PDUSessionResourceNotify = 30 + PrivateMessage = 31 + PWSCancel = 32 + PWSFailureIndication = 33 + PWSRestartIndication = 34 + RANConfigurationUpdate = 35 + RerouteNASRequest = 36 + RRCInactiveTransitionReport = 37 + TraceFailureIndication = 38 + TraceStart = 39 + UEContextModification = 40 + UEContextRelease = 41 + UEContextReleaseRequest = 42 + UERadioCapabilityCheck = 43 + UERadioCapabilityInfoIndication = 44 + UETNLABindingRelease = 45 + UplinkNASTransport = 46 + UplinkNonUEAssociatedNRPPaTransport = 47 + UplinkRANConfigurationTransfer = 48 + UplinkRANStatusTransfer = 49 + UplinkUEAssociatedNRPPaTransport = 50 + WriteReplaceWarning = 51 + SecondaryRATDataUsageReport = 52 + UplinkRIMInformationTransfer = 53 + DownlinkRIMInformationTransfer = 54 + RetrieveUEInformation = 55 + UEInformationTransfer = 56 + RANCPRelocationIndication = 57 + UEContextResume = 58 + UEContextSuspend = 59 + UERadioCapabilityIDMapping = 60 + HandoverSuccess = 61 + UplinkRANEarlyStatusTransfer = 62 + DownlinkRANEarlyStatusTransfer = 63 + AMFCPRelocationIndication = 64 + ConnectionEstablishmentIndication = 65 + BroadcastSessionModification = 66 + BroadcastSessionRelease = 67 + BroadcastSessionSetup = 68 + DistributionSetup = 69 + DistributionRelease = 70 + MulticastSessionActivation = 71 + MulticastSessionDeactivation = 72 + MulticastSessionUpdate = 73 + MulticastGroupPaging = 74 + BroadcastSessionReleaseRequired = 75 + TimingSynchronisationStatus = 76 + TimingSynchronisationStatusReport = 77 + MTCommunicationHandling = 78 + RANPagingRequest = 79 + BroadcastSessionTransport = 80 + + @staticmethod + def get(key: 'int | str | ProcedureCode', default: 'int' = -1) -> 'ProcedureCode': + """Backport support for original codes. + + Args: + key: Key to get enum item. + default: Default value if not found. + + :meta private: + """ + if isinstance(key, ProcedureCode): + return key + if isinstance(key, int): + return ProcedureCode(key) + if key not in ProcedureCode._member_map_: # pylint: disable=no-member + return extend_enum(ProcedureCode, key, default) + return ProcedureCode[key] # type: ignore[misc] + + @classmethod + def _missing_(cls, value: 'int') -> 'ProcedureCode': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + """ + if not (isinstance(value, int) and 0 <= value <= 255): + raise ValueError('%r is not a valid %s' % (value, cls.__name__)) + return extend_enum(cls, 'Unassigned_%d' % value, value) + + +class ProtocolIE(IntEnum): + """[ProtocolIE-ID] NGAP protocol IE identifiers, 3GPP TS 38.413. + + An IE's ID name and the name of the open type its value is keyed under + differ in places -- IE 21 is ``id-DefaultPagingDRX`` but its value arrives + keyed ``PagingDRX`` -- so + :attr:`IE.type ` carries + the latter alongside this. + + """ + + AllowedNSSAI = 0 + AMFName = 1 + AMFOverloadResponse = 2 + AMFSetID = 3 + AMF_TNLAssociationFailedToSetupList = 4 + AMF_TNLAssociationSetupList = 5 + AMF_TNLAssociationToAddList = 6 + AMF_TNLAssociationToRemoveList = 7 + AMF_TNLAssociationToUpdateList = 8 + AMFTrafficLoadReductionIndication = 9 + AMF_UE_NGAP_ID = 10 + AssistanceDataForPaging = 11 + BroadcastCancelledAreaList = 12 + BroadcastCompletedAreaList = 13 + CancelAllWarningMessages = 14 + Cause = 15 + CellIDListForRestart = 16 + ConcurrentWarningMessageInd = 17 + CoreNetworkAssistanceInformationForInactive = 18 + CriticalityDiagnostics = 19 + DataCodingScheme = 20 + DefaultPagingDRX = 21 + DirectForwardingPathAvailability = 22 + EmergencyAreaIDListForRestart = 23 + EmergencyFallbackIndicator = 24 + EUTRA_CGI = 25 + FiveG_S_TMSI = 26 + GlobalRANNodeID = 27 + GUAMI = 28 + HandoverType = 29 + IMSVoiceSupportIndicator = 30 + IndexToRFSP = 31 + InfoOnRecommendedCellsAndRANNodesForPaging = 32 + LocationReportingRequestType = 33 + MaskedIMEISV = 34 + MessageIdentifier = 35 + MobilityRestrictionList = 36 + NASC = 37 + NAS_PDU = 38 + NASSecurityParametersFromNGRAN = 39 + NewAMF_UE_NGAP_ID = 40 + NewSecurityContextInd = 41 + NGAP_Message = 42 + NGRAN_CGI = 43 + NGRANTraceID = 44 + NR_CGI = 45 + NRPPa_PDU = 46 + NumberOfBroadcastsRequested = 47 + OldAMF = 48 + OverloadStartNSSAIList = 49 + PagingDRX = 50 + PagingOrigin = 51 + PagingPriority = 52 + PDUSessionResourceAdmittedList = 53 + PDUSessionResourceFailedToModifyListModRes = 54 + PDUSessionResourceFailedToSetupListCxtRes = 55 + PDUSessionResourceFailedToSetupListHOAck = 56 + PDUSessionResourceFailedToSetupListPSReq = 57 + PDUSessionResourceFailedToSetupListSURes = 58 + PDUSessionResourceHandoverList = 59 + PDUSessionResourceListCxtRelCpl = 60 + PDUSessionResourceListHORqd = 61 + PDUSessionResourceModifyListModCfm = 62 + PDUSessionResourceModifyListModInd = 63 + PDUSessionResourceModifyListModReq = 64 + PDUSessionResourceModifyListModRes = 65 + PDUSessionResourceNotifyList = 66 + PDUSessionResourceReleasedListNot = 67 + PDUSessionResourceReleasedListPSAck = 68 + PDUSessionResourceReleasedListPSFail = 69 + PDUSessionResourceReleasedListRelRes = 70 + PDUSessionResourceSetupListCxtReq = 71 + PDUSessionResourceSetupListCxtRes = 72 + PDUSessionResourceSetupListHOReq = 73 + PDUSessionResourceSetupListSUReq = 74 + PDUSessionResourceSetupListSURes = 75 + PDUSessionResourceToBeSwitchedDLList = 76 + PDUSessionResourceSwitchedList = 77 + PDUSessionResourceToReleaseListHOCmd = 78 + PDUSessionResourceToReleaseListRelCmd = 79 + PLMNSupportList = 80 + PWSFailedCellIDList = 81 + RANNodeName = 82 + RANPagingPriority = 83 + RANStatusTransfer_TransparentContainer = 84 + RAN_UE_NGAP_ID = 85 + RelativeAMFCapacity = 86 + RepetitionPeriod = 87 + ResetType = 88 + RoutingID = 89 + RRCEstablishmentCause = 90 + RRCInactiveTransitionReportRequest = 91 + RRCState = 92 + SecurityContext = 93 + SecurityKey = 94 + SerialNumber = 95 + ServedGUAMIList = 96 + SliceSupportList = 97 + SONConfigurationTransferDL = 98 + SONConfigurationTransferUL = 99 + SourceAMF_UE_NGAP_ID = 100 + SourceToTarget_TransparentContainer = 101 + SupportedTAList = 102 + TAIListForPaging = 103 + TAIListForRestart = 104 + TargetID = 105 + TargetToSource_TransparentContainer = 106 + TimeToWait = 107 + TraceActivation = 108 + TraceCollectionEntityIPAddress = 109 + UEAggregateMaximumBitRate = 110 + UE_associatedLogicalNG_connectionList = 111 + UEContextRequest = 112 + UE_NGAP_IDs = 114 + UEPagingIdentity = 115 + UEPresenceInAreaOfInterestList = 116 + UERadioCapability = 117 + UERadioCapabilityForPaging = 118 + UESecurityCapabilities = 119 + UnavailableGUAMIList = 120 + UserLocationInformation = 121 + WarningAreaList = 122 + WarningMessageContents = 123 + WarningSecurityInfo = 124 + WarningType = 125 + AdditionalUL_NGU_UP_TNLInformation = 126 + DataForwardingNotPossible = 127 + DL_NGU_UP_TNLInformation = 128 + NetworkInstance = 129 + PDUSessionAggregateMaximumBitRate = 130 + PDUSessionResourceFailedToModifyListModCfm = 131 + PDUSessionResourceFailedToSetupListCxtFail = 132 + PDUSessionResourceListCxtRelReq = 133 + PDUSessionType = 134 + QosFlowAddOrModifyRequestList = 135 + QosFlowSetupRequestList = 136 + QosFlowToReleaseList = 137 + SecurityIndication = 138 + UL_NGU_UP_TNLInformation = 139 + UL_NGU_UP_TNLModifyList = 140 + WarningAreaCoordinates = 141 + PDUSessionResourceSecondaryRATUsageList = 142 + HandoverFlag = 143 + SecondaryRATUsageInformation = 144 + PDUSessionResourceReleaseResponseTransfer = 145 + RedirectionVoiceFallback = 146 + UERetentionInformation = 147 + S_NSSAI = 148 + PSCellInformation = 149 + LastEUTRAN_PLMNIdentity = 150 + MaximumIntegrityProtectedDataRate_DL = 151 + AdditionalDLForwardingUPTNLInformation = 152 + AdditionalDLUPTNLInformationForHOList = 153 + AdditionalNGU_UP_TNLInformation = 154 + AdditionalDLQosFlowPerTNLInformation = 155 + SecurityResult = 156 + ENDC_SONConfigurationTransferDL = 157 + ENDC_SONConfigurationTransferUL = 158 + OldAssociatedQosFlowList_ULendmarkerexpected = 159 + CNTypeRestrictionsForEquivalent = 160 + CNTypeRestrictionsForServing = 161 + NewGUAMI = 162 + ULForwarding = 163 + ULForwardingUP_TNLInformation = 164 + CNAssistedRANTuning = 165 + CommonNetworkInstance = 166 + NGRAN_TNLAssociationToRemoveList = 167 + TNLAssociationTransportLayerAddressNGRAN = 168 + EndpointIPAddressAndPort = 169 + LocationReportingAdditionalInfo = 170 + SourceToTarget_AMFInformationReroute = 171 + AdditionalULForwardingUPTNLInformation = 172 + SCTP_TLAs = 173 + SelectedPLMNIdentity = 174 + RIMInformationTransfer = 175 + GUAMIType = 176 + SRVCCOperationPossible = 177 + TargetRNC_ID = 178 + RAT_Information = 179 + ExtendedRATRestrictionInformation = 180 + QosMonitoringRequest = 181 + SgNB_UE_X2AP_ID = 182 + AdditionalRedundantDL_NGU_UP_TNLInformation = 183 + AdditionalRedundantDLQosFlowPerTNLInformation = 184 + AdditionalRedundantNGU_UP_TNLInformation = 185 + AdditionalRedundantUL_NGU_UP_TNLInformation = 186 + CNPacketDelayBudgetDL = 187 + CNPacketDelayBudgetUL = 188 + ExtendedPacketDelayBudget = 189 + RedundantCommonNetworkInstance = 190 + RedundantDL_NGU_TNLInformationReused = 191 + RedundantDL_NGU_UP_TNLInformation = 192 + RedundantDLQosFlowPerTNLInformation = 193 + RedundantQosFlowIndicator = 194 + RedundantUL_NGU_UP_TNLInformation = 195 + TSCTrafficCharacteristics = 196 + RedundantPDUSessionInformation = 197 + UsedRSNInformation = 198 + IAB_Authorized = 199 + IAB_Supported = 200 + IABNodeIndication = 201 + NB_IoT_PagingDRX = 202 + NB_IoT_Paging_eDRXInfo = 203 + NB_IoT_DefaultPagingDRX = 204 + Enhanced_CoverageRestriction = 205 + Extended_ConnectedTime = 206 + PagingAssisDataforCEcapabUE = 207 + WUS_Assistance_Information = 208 + UE_DifferentiationInfo = 209 + NB_IoT_UEPriority = 210 + UL_CP_SecurityInformation = 211 + DL_CP_SecurityInformation = 212 + TAI = 213 + UERadioCapabilityForPagingOfNB_IoT = 214 + LTEV2XServicesAuthorized = 215 + NRV2XServicesAuthorized = 216 + LTEUESidelinkAggregateMaximumBitrate = 217 + NRUESidelinkAggregateMaximumBitrate = 218 + PC5QoSParameters = 219 + AlternativeQoSParaSetList = 220 + CurrentQoSParaSetIndex = 221 + CEmodeBrestricted = 222 + EUTRA_PagingeDRXInformation = 223 + CEmodeBSupport_Indicator = 224 + LTEM_Indication = 225 + EndIndication = 226 + EDT_Session = 227 + UECapabilityInfoRequest = 228 + PDUSessionResourceFailedToResumeListRESReq = 229 + PDUSessionResourceFailedToResumeListRESRes = 230 + PDUSessionResourceSuspendListSUSReq = 231 + PDUSessionResourceResumeListRESReq = 232 + PDUSessionResourceResumeListRESRes = 233 + UE_UP_CIoT_Support = 234 + Suspend_Request_Indication = 235 + Suspend_Response_Indication = 236 + RRC_Resume_Cause = 237 + RGLevelWirelineAccessCharacteristics = 238 + W_AGFIdentityInformation = 239 + GlobalTNGF_ID = 240 + GlobalTWIF_ID = 241 + GlobalW_AGF_ID = 242 + UserLocationInformationW_AGF = 243 + UserLocationInformationTNGF = 244 + AuthenticatedIndication = 245 + TNGFIdentityInformation = 246 + TWIFIdentityInformation = 247 + UserLocationInformationTWIF = 248 + DataForwardingResponseERABList = 249 + IntersystemSONConfigurationTransferDL = 250 + IntersystemSONConfigurationTransferUL = 251 + SONInformationReport = 252 + UEHistoryInformationFromTheUE = 253 + ManagementBasedMDTPLMNList = 254 + MDTConfiguration = 255 + PrivacyIndicator = 256 + TraceCollectionEntityURI = 257 + NPN_Support = 258 + NPN_AccessInformation = 259 + NPN_PagingAssistanceInformation = 260 + NPN_MobilityInformation = 261 + TargettoSource_Failure_TransparentContainer = 262 + NID = 263 + UERadioCapabilityID = 264 + UERadioCapability_EUTRA_Format = 265 + DAPSRequestInfo = 266 + DAPSResponseInfoList = 267 + EarlyStatusTransfer_TransparentContainer = 268 + NotifySourceNGRANNode = 269 + ExtendedSliceSupportList = 270 + ExtendedTAISliceSupportList = 271 + ConfiguredTACIndication = 272 + Extended_RANNodeName = 273 + Extended_AMFName = 274 + GlobalCable_ID = 275 + QosMonitoringReportingFrequency = 276 + QosFlowParametersList = 277 + QosFlowFeedbackList = 278 + BurstArrivalTimeDownlink = 279 + ExtendedUEIdentityIndexValue = 280 + PduSessionExpectedUEActivityBehaviour = 281 + MicoAllPLMN = 282 + QosFlowFailedToSetupList = 283 + SourceTNLAddrInfo = 284 + ExtendedReportIntervalMDT = 285 + SourceNodeID = 286 + NRNTNTAIInformation = 287 + UEContextReferenceAtSource = 288 + LastVisitedPSCellList = 289 + IntersystemSONInformationRequest = 290 + IntersystemSONInformationReply = 291 + EnergySavingIndication = 292 + IntersystemResourceStatusUpdate = 293 + SuccessfulHandoverReportList = 294 + MBS_AreaSessionID = 295 + MBS_QoSFlowsToBeSetupList = 296 + MBS_QoSFlowsToBeSetupModList = 297 + MBS_ServiceArea = 298 + MBS_SessionID = 299 + MBS_DistributionReleaseRequestTransfer = 300 + MBS_DistributionSetupRequestTransfer = 301 + MBS_DistributionSetupResponseTransfer = 302 + MBS_DistributionSetupUnsuccessfulTransfer = 303 + MulticastSessionActivationRequestTransfer = 304 + MulticastSessionDeactivationRequestTransfer = 305 + MulticastSessionUpdateRequestTransfer = 306 + MulticastGroupPagingAreaList = 307 + MBS_SupportIndicator = 309 + MBSSessionFailedtoSetupList = 310 + MBSSessionFailedtoSetuporModifyList = 311 + MBSSessionSetupResponseList = 312 + MBSSessionSetuporModifyResponseList = 313 + MBSSessionSetupFailureTransfer = 314 + MBSSessionSetupRequestTransfer = 315 + MBSSessionSetupResponseTransfer = 316 + MBSSessionToReleaseList = 317 + MBSSessionSetupRequestList = 318 + MBSSessionSetuporModifyRequestList = 319 + MBS_ActiveSessionInformation_SourcetoTargetList = 323 + MBS_ActiveSessionInformation_TargettoSourceList = 324 + OnboardingSupport = 325 + TimeSyncAssistanceInfo = 326 + SurvivalTime = 327 + QMCConfigInfo = 328 + QMCDeactivation = 329 + PDUSessionPairID = 331 + NR_PagingeDRXInformation = 332 + RedCapIndication = 333 + TargetNSSAIInformation = 334 + UESliceMaximumBitRateList = 335 + M4ReportAmount = 336 + M5ReportAmount = 337 + M6ReportAmount = 338 + M7ReportAmount = 339 + IncludeBeamMeasurementsIndication = 340 + ExcessPacketDelayThresholdConfiguration = 341 + PagingCause = 342 + PagingCauseIndicationForVoiceService = 343 + PEIPSassistanceInformation = 344 + FiveG_ProSeAuthorized = 345 + FiveG_ProSeUEPC5AggregateMaximumBitRate = 346 + FiveG_ProSePC5QoSParameters = 347 + MBSSessionModificationFailureTransfer = 348 + MBSSessionModificationRequestTransfer = 349 + MBSSessionModificationResponseTransfer = 350 + MBS_QoSFlowToReleaseList = 351 + MBS_SessionTNLInfo5GC = 352 + TAINSAGSupportList = 353 + SourceNodeTNLAddrInfo = 354 + NGAPIESupportInformationRequestList = 355 + NGAPIESupportInformationResponseList = 356 + MBS_SessionFSAIDList = 357 + MBSSessionReleaseResponseTransfer = 358 + ManagementBasedMDTPLMNModificationList = 359 + EarlyMeasurement = 360 + BeamMeasurementsReportConfiguration = 361 + HFCNode_ID_new = 362 + GlobalCable_ID_new = 363 + TargetHomeENB_ID = 364 + HashedUEIdentityIndexValue = 365 + ExtendedMobilityInformation = 366 + NetworkControlledRepeaterAuthorized = 367 + AdditionalCancelledlocationReportingReferenceIDList = 368 + Selected_Target_SNPN_Identity = 369 + EquivalentSNPNsList = 370 + SelectedNID = 371 + SupportedUETypeList = 372 + AerialUEsubscriptionInformation = 373 + NR_A2X_ServicesAuthorized = 374 + LTE_A2X_ServicesAuthorized = 375 + NR_A2X_UE_PC5_AggregateMaximumBitRate = 376 + LTE_A2X_UE_PC5_AggregateMaximumBitRate = 377 + A2X_PC5_QoS_Parameters = 378 + FiveGProSeLayer2Multipath = 379 + FiveGProSeLayer2UEtoUERelay = 380 + FiveGProSeLayer2UEtoUERemote = 381 + CandidateRelayUEInformationList = 382 + SuccessfulPSCellChangeReportList = 383 + IntersystemMobilityFailureforVoiceFallback = 384 + TargetCellCRNTI = 385 + TimeSinceFailure = 386 + RANTimingSynchronisationStatusInfo = 387 + RAN_TSSRequestType = 388 + RAN_TSSScope = 389 + ClockQualityReportingControlInfo = 390 + RANfeedbacktype = 391 + QoSFlowTSCList = 392 + TSCTrafficCharacteristicsFeedback = 393 + DownlinkTLContainer = 394 + UplinkTLContainer = 395 + ANPacketDelayBudgetUL = 396 + QosFlowAdditionalInfoList = 397 + AssistanceInformationQoE_Meas = 398 + MBSCommServiceType = 399 + MobileIAB_Authorized = 400 + MobileIAB_MTUserLocationInformation = 401 + MobileIABNodeIndication = 402 + NoPDUSessionIndication = 403 + MobileIAB_Supported = 404 + CN_MT_CommunicationHandling = 405 + FiveGCAction = 406 + PagingPolicyDifferentiation = 407 + DL_Signalling = 408 + PNI_NPN_AreaScopeofMDT = 409 + PNI_NPNBasedMDT = 410 + SNPN_CellBasedMDT = 411 + SNPN_TAIBasedMDT = 412 + SNPN_BasedMDT = 413 + Partially_Allowed_NSSAI = 414 + AssociatedSessionID = 415 + MBS_AssistanceInformation = 416 + BroadcastTransportFailureTransfer = 417 + BroadcastTransportRequestTransfer = 418 + BroadcastTransportResponseTransfer = 419 + TimeBasedHandoverInformation = 420 + DLDiscarding = 421 + PDUsetQoSParameters = 422 + PDUSetbasedHandlingIndicator = 423 + N6JitterInformation = 424 + ECNMarkingorCongestionInformationReportingRequest = 425 + ECNMarkingorCongestionInformationReportingStatus = 426 + ERedCapIndication = 427 + XrDeviceWith2Rx = 428 + UserPlaneErrorIndicator = 429 + SLPositioningRangingServiceInfo = 430 + PDUSessionListMTCommHReq = 431 + MaximumDataBurstVolume = 432 + MN_only_MDT_collection = 433 + MBS_NGUFailureIndication = 434 + UserPlaneFailureIndication = 435 + UserPlaneFailureIndicationReport = 436 + SourceSN_to_TargetSN_QMCInfo = 437 + QoERVQoEReportingPaths = 438 + UserLocationInformationN3IWF_without_PortNumber = 439 + AUN3DeviceAccessInfo = 440 + TAIMBSSupportList = 441 + ExtendedBackupAMFName = 442 + ExtendedOldAMF = 443 + + @staticmethod + def get(key: 'int | str | ProtocolIE', default: 'int' = -1) -> 'ProtocolIE': + """Backport support for original codes. + + Args: + key: Key to get enum item. + default: Default value if not found. + + :meta private: + """ + if isinstance(key, ProtocolIE): + return key + if isinstance(key, int): + return ProtocolIE(key) + if key not in ProtocolIE._member_map_: # pylint: disable=no-member + return extend_enum(ProtocolIE, key, default) + return ProtocolIE[key] # type: ignore[misc] + + @classmethod + def _missing_(cls, value: 'int') -> 'ProtocolIE': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + """ + if not (isinstance(value, int) and 0 <= value <= 65535): + raise ValueError('%r is not a valid %s' % (value, cls.__name__)) + return extend_enum(cls, 'Unassigned_%d' % value, value) + + +############################################################################## +# Value tree conversion. +############################################################################## + + +def _convert(value: 'Any') -> 'Any': + """Convert a |pycrate|_ decoded value into :mod:`pcapkit` data models. + + Args: + value: A node of the tree returned by ``NGAP_PDU.get_val()``. + + Returns: + The same tree, with mappings as + :class:`~pcapkit.protocols.data.application.ngap.Sequence`, ``CHOICE`` + pairs as :class:`~pcapkit.protocols.data.application.ngap.Choice`, and + ``BIT STRING`` pairs as + :class:`~pcapkit.protocols.data.application.ngap.BitString`. + + Notes: + The two 2-tuple shapes are told apart by their first element, which is + unambiguous: |pycrate|_ spells a ``CHOICE`` as ``(name, value)`` with a + :obj:`str` name and a ``BIT STRING`` as ``(bits, length)`` with two + :obj:`int`. Any other tuple is passed through with its members + converted, so an ASN.1 construct not listed above degrades to its own + shape rather than being mangled into one of these. + + """ + if isinstance(value, dict): + return Data_Sequence({key: _convert(val) for key, val in value.items()}) + if isinstance(value, list): + return [_convert(item) for item in value] + if isinstance(value, tuple): + if len(value) == 2: + if isinstance(value[0], str): + return Data_Choice(name=value[0], value=_convert(value[1])) + if isinstance(value[0], int) and isinstance(value[1], int): + return Data_BitString(value=value[0], length=value[1]) + return tuple(_convert(item) for item in value) + return value + + +def _revert(value: 'Any') -> 'Any': + """Convert :mod:`pcapkit` data models back into a |pycrate|_ value. + + Args: + value: A node of a tree produced by :func:`_convert`, or the equivalent + plain Python value. + + Returns: + The same tree in the shape ``NGAP_PDU.set_val()`` expects. + + Notes: + :class:`~pcapkit.protocols.data.application.ngap.Choice` and + :class:`~pcapkit.protocols.data.application.ngap.BitString` are tested + before :class:`~pcapkit.corekit.infoclass.Info`, since they are + subclasses of it and would otherwise be reverted to mappings. + + """ + if isinstance(value, Data_Choice): + return (value.name, _revert(value.value)) + if isinstance(value, Data_BitString): + return (value.value, value.length) + if isinstance(value, Info): + return {key: _revert(value[key]) for key in value} + if isinstance(value, list): + return [_revert(item) for item in value] + if isinstance(value, tuple): + return tuple(_revert(item) for item in value) + return value + + +############################################################################## +# Protocol. +############################################################################## + + +class NGAP(Application[Data_NGAP, Schema_NGAP], + data=Data_NGAP, schema=Schema_NGAP): + """This class implements NG Application Protocol.""" + + ########################################################################## + # Properties. + ########################################################################## + + @property + def name(self) -> 'Literal["NG Application Protocol"]': + """Name of current protocol.""" + return 'NG Application Protocol' + + @property + def length(self) -> 'int': + """Header length of current protocol. + + NGAP prefixes its payload with nothing and carries no next layer, so the + whole ``NGAP-PDU`` *is* the header and this is its length. That is not + :meth:`self.__length_hint__ `, which reports the + four-octet prefix common to every PDU rather than this PDU's size. + + """ + return len(self.__header__.data) + + ########################################################################## + # Methods. + ########################################################################## + + def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_NGAP': # pylint: disable=unused-argument + """Read NG Application Protocol (NGAP). + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If |pycrate|_ is not installed, or if the payload is + not a well formed aligned PER ``NGAP-PDU``. + + """ + if length is None: + length = len(self) + schema = self.__header__ + + pdu = load_pycrate() + if pdu is None: + raise ProtocolError('NGAP: decoding needs the optional "pycrate" dependency, ' + 'which is not installed; pip install pypcapkit[NGAP]') + + with _PDU_LOCK: + try: + # NOTE: `reset_val()` must not be called here, tempting though it + # looks before a decode into a reused object. It walks all 318 + # submodules of the compiled specification and costs ~105 ms, + # some 700x the ~0.15 ms of the decode itself; and `from_aper()` + # overwrites the stored value outright, so it would buy nothing + # even if it were free. + pdu.from_aper(schema.data) + kind, body = pdu.get_val() + + message, message_body = body['value'] + + # NOTE: converted while the lock is held, because `get_val()` + # returns the decoder's own containers rather than copies -- the + # next decode through this module-level object rewrites them. + value = _convert(message_body) + + ies = [] # type: list[Data_IE] + for ie in value.get('protocolIEs') or (): + choice = ie['value'] + ies.append(Data_IE( + id=ProtocolIE.get(ie['id']), + criticality=Criticality.get(ie['criticality']), + type=choice.name, + value=choice.value, + )) + + ngap = Data_NGAP( + kind=PDUKind(kind), + procedure=ProcedureCode.get(body['procedureCode']), + criticality=Criticality.get(body['criticality']), + message=message, + ies=tuple(ies), + value=value, + ) + except ProtocolError: + # NOTE: a ProtocolError raised inside this block is already + # specific about what went wrong, so it must not be caught below + # and relabelled "malformed NGAP-PDU". Nothing here raises one + # today; this is what keeps that true of the next edit. + raise + except Exception as exc: + # NOTE: `Exception` rather than something narrower on purpose. + # pycrate signals a bad encoding from two unrelated hierarchies + # -- `pycrate_asn1rt.err.ASN1Err` and `pycrate_core.charpy. + # CharpyErr` -- and a truncated payload can also surface as + # `KeyError` or `IndexError` from inside the codec. None of them + # is a pcapkit exception, and letting one escape would make an + # NGAP packet fail differently from every other protocol. + raise ProtocolError(f'NGAP: malformed NGAP-PDU: {exc}') from exc + return ngap + + def make(self, + kind: 'PDUKind | str' = PDUKind.INITIATING_MESSAGE, + procedure: 'Optional[ProcedureCode | int]' = None, + criticality: 'Criticality | int | str' = Criticality.reject, + message: 'Optional[str]' = None, + value: 'Any' = None, + data: 'Optional[bytes]' = None, + **kwargs: 'Any') -> 'Schema_NGAP': + """Make (construct) packet data. + + Args: + kind: Which ``NGAP-PDU`` alternative to construct. + procedure: Procedure code. + criticality: Criticality of the procedure. + message: Name of the message type, e.g. ``NGSetupRequest``. + value: Message body, either as + :class:`~pcapkit.protocols.data.application.ngap.Sequence` from + a previous :meth:`read` or as the plain Python value + |pycrate|_ expects. + data: Pre-encoded ``NGAP-PDU``. When given, it is used verbatim and + every other argument is ignored, which is also the only path + that does not need |pycrate|_. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed packet data. + + Raises: + ProtocolError: If ``data`` is not given and either ``procedure`` or + ``message`` is missing, if |pycrate|_ is not installed, or if + the arguments do not describe a message the specification can + encode. + + """ + if data is not None: + return Schema_NGAP(data=bytes(data)) + + if procedure is None or message is None: + raise ProtocolError("NGAP: 'procedure' and 'message' are required " + "when 'data' is not given") + + pdu = load_pycrate() + if pdu is None: + raise ProtocolError('NGAP: encoding needs the optional "pycrate" dependency, ' + 'which is not installed; pip install pypcapkit[NGAP]') + + with _PDU_LOCK: + try: + pdu.set_val((PDUKind(kind).value, { + 'procedureCode': int(ProcedureCode.get(procedure)), + 'criticality': Criticality.get(criticality).name, + 'value': (message, _revert(value)), + })) + encoded = pdu.to_aper() + except Exception as exc: + # NOTE: same reasoning as the read path -- pycrate's encoder + # raises from its own hierarchies, none of them pcapkit's. + raise ProtocolError(f'NGAP: cannot encode NGAP-PDU: {exc}') from exc + return Schema_NGAP(data=encoded) + + ########################################################################## + # Data models. + ########################################################################## + + def __length_hint__(self) -> 'Literal[4]': + """Return an estimated length for the object. + + Every ``NGAP-PDU`` opens with the same four octets under aligned PER -- + the ``CHOICE`` index, the procedure code, the criticality, and the + first octet of the open type's length determinant -- so four is the + fixed prefix NGAP has in place of a header. It is not a minimum PDU + length: the smallest complete ``NGAP-PDU`` measured here, a message + whose ``protocolIEs`` list is empty, is seven octets. + + """ + return 4 + + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_NGAP') -> '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 { + 'kind': data.kind, + 'procedure': data.procedure, + 'criticality': data.criticality, + 'message': data.message, + 'value': data.value, + 'data': None, + } diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index c5da5bb591..1224924c0c 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -192,6 +192,10 @@ 'FTP', 'FTP_Request', 'FTP_Response', + # NG Application Protocol + 'NGAP', + 'NGAP_IE', 'NGAP_Choice', 'NGAP_BitString', 'NGAP_Sequence', + # Hypertext Transfer Protocol (HTTP/1.*) 'HTTPv1', 'HTTPv1_Header', diff --git a/pcapkit/protocols/data/application/__init__.py b/pcapkit/protocols/data/application/__init__.py index f653342a29..620823252b 100644 --- a/pcapkit/protocols/data/application/__init__.py +++ b/pcapkit/protocols/data/application/__init__.py @@ -36,6 +36,13 @@ from pcapkit.protocols.data.application.httpv2 import UnassignedFrame as HTTPv2_UnassignedFrame from pcapkit.protocols.data.application.httpv2 import WindowUpdateFrame as HTTPv2_WindowUpdateFrame +# NG Application Protocol +from pcapkit.protocols.data.application.ngap import IE as NGAP_IE +from pcapkit.protocols.data.application.ngap import NGAP +from pcapkit.protocols.data.application.ngap import BitString as NGAP_BitString +from pcapkit.protocols.data.application.ngap import Choice as NGAP_Choice +from pcapkit.protocols.data.application.ngap import Sequence as NGAP_Sequence + __all__ = [ # File Transfer Protocol 'FTP', @@ -54,4 +61,8 @@ 'HTTPv2_UnassignedFrame', 'HTTPv2_DataFrame', 'HTTPv2_HeadersFrame', 'HTTPv2_PriorityFrame', 'HTTPv2_RSTStreamFrame', 'HTTPv2_SettingsFrame', 'HTTPv2_PushPromiseFrame', 'HTTPv2_PingFrame', 'HTTPv2_GoawayFrame', 'HTTPv2_WindowUpdateFrame', 'HTTPv2_ContinuationFrame', + + # NG Application Protocol + 'NGAP', + 'NGAP_IE', 'NGAP_Choice', 'NGAP_BitString', 'NGAP_Sequence', ] diff --git a/pcapkit/protocols/data/application/ngap.py b/pcapkit/protocols/data/application/ngap.py new file mode 100644 index 0000000000..84b7c7894e --- /dev/null +++ b/pcapkit/protocols/data/application/ngap.py @@ -0,0 +1,123 @@ +# -*- coding: utf-8 -*- +"""data models for NGAP protocol""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.infoclass import info_final +from pcapkit.protocols.data.data import Data +from pcapkit.protocols.data.protocol import Protocol + +if TYPE_CHECKING: + from typing import Any + + from pcapkit.protocols.application.ngap import Criticality, PDUKind, ProcedureCode, ProtocolIE + +__all__ = [ + 'NGAP', + 'IE', 'Choice', 'BitString', 'Sequence', +] + + +@info_final +class Sequence(Data): + """Data model for an ASN.1 ``SEQUENCE``, ``SET`` or ``SEQUENCE OF`` member. + + Fields are whatever the specification names them, so this model carries no + fixed annotations: it is populated from the decoded value tree. ASN.1 + identifiers are hyphenated where Python identifiers cannot be, e.g. + ``gNB-ID``, so such fields are reachable by subscription + (``seq['gNB-ID']``) rather than by attribute access. + + """ + + +@info_final +class BitString(Data): + """Data model for an ASN.1 ``BIT STRING``. + + A bit string is not a whole number of octets, so its length is carried + alongside its value rather than being implied by it -- ``gNB-ID`` is a + 22-to-32-bit field, and ``(0x000102, 24)`` and ``(0x000102, 32)`` are + different identifiers. + + """ + + #: Bits, as a big-endian unsigned integer. + value: 'int' + #: Number of significant bits in :attr:`value`. + length: 'int' + + if TYPE_CHECKING: + def __init__(self, value: 'int', length: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements + + +@info_final +class Choice(Data): + """Data model for an ASN.1 ``CHOICE`` alternative or open type. + + Both the selected alternative's name and its value are kept, since the name + is the only thing that says *which* of the alternatives was sent -- an + ``NGAP-PDU`` carrying ``('globalGNB-ID', ...)`` and one carrying + ``('globalNgENB-ID', ...)`` are otherwise indistinguishable once the value + has been converted. + + """ + + #: Name of the selected alternative, as spelled in 3GPP TS 38.413. + name: 'str' + #: Value of the selected alternative. + value: 'Any' + + if TYPE_CHECKING: + def __init__(self, name: 'str', value: 'Any') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements + + +@info_final +class IE(Data): + """Data model for one NGAP protocol information element.""" + + #: Protocol IE ID. + id: 'ProtocolIE' + #: Criticality, i.e. what a receiver must do when it does not understand + #: :attr:`id`. + criticality: 'Criticality' + #: Name of the IE's open type, as spelled in 3GPP TS 38.413. This is not + #: always :attr:`id`'s own spelling -- IE 21 is ``id-DefaultPagingDRX`` + #: but its value is keyed ``PagingDRX``. + type: 'str' + #: Value of the IE. + value: 'Any' + + if TYPE_CHECKING: + def __init__(self, id: 'ProtocolIE', criticality: 'Criticality', type: 'str', value: 'Any') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long + + +@info_final +class NGAP(Protocol): + """Data model for NGAP protocol. + + The three ``NGAP-PDU`` alternatives -- ``initiatingMessage``, + ``successfulOutcome`` and ``unsuccessfulOutcome`` -- carry an identical + field set and are distinguished by :attr:`kind` rather than by three + near-identical models. + + """ + + #: Which of the three ``NGAP-PDU`` alternatives this is. + kind: 'PDUKind' + #: Procedure code. + procedure: 'ProcedureCode' + #: Criticality of the procedure. + criticality: 'Criticality' + #: Name of the message type, e.g. ``NGSetupRequest``. + message: 'str' + #: Protocol IEs of the message, in the order they were encoded. Empty for + #: the few messages that carry no ``protocolIEs`` field. + ies: 'tuple[IE, ...]' + #: The message body, converted in full. :attr:`ies` is a flattened view of + #: its ``protocolIEs`` field, so anything not surfaced above is reachable + #: here. + value: 'Sequence' + + if TYPE_CHECKING: + def __init__(self, kind: 'PDUKind', procedure: 'ProcedureCode', criticality: 'Criticality', message: 'str', ies: 'tuple[IE, ...]', value: 'Sequence') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index c567dca89c..19108c3225 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -123,6 +123,7 @@ # Application Layer Protocols 'FTP', + 'NGAP', 'HTTPv1', 'HTTPv2', 'HTTPv2_FrameType', diff --git a/pcapkit/protocols/schema/application/__init__.py b/pcapkit/protocols/schema/application/__init__.py index 91282c9ff0..a1f010f9f2 100644 --- a/pcapkit/protocols/schema/application/__init__.py +++ b/pcapkit/protocols/schema/application/__init__.py @@ -24,6 +24,9 @@ from pcapkit.protocols.schema.application.httpv2 import \ WindowUpdateFrame as HTTPv2_WindowUpdateFrame +# NG Application Protocol +from pcapkit.protocols.schema.application.ngap import NGAP + __all__ = [ # File Transfer Protocol 'FTP', @@ -37,4 +40,7 @@ 'HTTPv2_UnassignedFrame', 'HTTPv2_DataFrame', 'HTTPv2_HeadersFrame', 'HTTPv2_PriorityFrame', 'HTTPv2_RSTStreamFrame', 'HTTPv2_SettingsFrame', 'HTTPv2_PushPromiseFrame', 'HTTPv2_PingFrame', 'HTTPv2_GoawayFrame', 'HTTPv2_WindowUpdateFrame', 'HTTPv2_ContinuationFrame', + + # NG Application Protocol + 'NGAP', ] diff --git a/pcapkit/protocols/schema/application/ngap.py b/pcapkit/protocols/schema/application/ngap.py new file mode 100644 index 0000000000..155bc37340 --- /dev/null +++ b/pcapkit/protocols/schema/application/ngap.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +# mypy: disable-error-code=assignment +"""header schema for NGAP protocol""" + +from typing import TYPE_CHECKING + +from pcapkit.corekit.fields.strings import BytesField +from pcapkit.protocols.schema.schema import Schema, schema_final + +__all__ = ['NGAP'] + + +@schema_final +class NGAP(Schema): + """Header schema for NGAP packet. + + NGAP has no header of its own: an SCTP DATA chunk whose payload protocol + identifier names NGAP carries exactly one aligned-PER-encoded ``NGAP-PDU`` + and nothing else, so there is no length field to read and no framing to + resolve. The whole payload is the encoding, and its structure only becomes + visible once the ASN.1 decoder has run. + + """ + + #: Aligned PER encoding of one ``NGAP-PDU``. + data: 'bytes' = BytesField(lambda pkt: pkt['__length__']) + + if TYPE_CHECKING: + def __init__(self, data: 'bytes') -> 'None': ... diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 5a8b8103ac..ac7bcc4663 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -220,7 +220,10 @@ class SCTP(Transport[Data_SCTP, Schema_SCTP], >>> SCTP.register(Enum_PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NG_Application_Protocol, NGAP) >>> SCTP.register(60, NGAP) # equivalent, PPID given as a plain integer - No PPID is registered by default. + Two PPIDs are registered by default, both to + :class:`~pcapkit.protocols.application.ngap.NGAP`: 60 + (``NG_Application_Protocol``) and 66 (``NGAP_over_DTLS_over_SCTP``). Every + other PPID resolves to :class:`~pcapkit.protocols.misc.raw.Raw`. This class currently supports parsing of the following SCTP chunks, which are directly mapped to the :class:`pcapkit.const.sctp.chunk.Chunk` @@ -343,7 +346,15 @@ class SCTP(Transport[Data_SCTP, Schema_SCTP], #: :class:`~pcapkit.protocols.transport.udp.UDP`. __proto__ = collections.defaultdict( lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), - {}, + { + # PPID 66 is NGAP wrapped in a DTLS record rather than a bare + # NGAP-PDU, and pcapkit implements no DTLS. It is registered anyway + # so that the PPID is *named*: the payload then fails in NGAP's own + # decoder and `beholder` degrades it to Raw, which is where an + # unregistered PPID would have left it regardless. + Enum_PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NG_Application_Protocol: ModuleDescriptor('pcapkit.protocols.application.ngap', 'NGAP'), # NGAP + Enum_PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NGAP_over_DTLS_over_SCTP: ModuleDescriptor('pcapkit.protocols.application.ngap', 'NGAP'), # NGAP over DTLS + }, ) # type: DefaultDict[int, ModuleDescriptor[Protocol] | Type[Protocol]] #: DefaultDict[Enum_Chunk, str | tuple[ChunkParser, ChunkConstructor]]: Chunk diff --git a/pyproject.toml b/pyproject.toml index 7a206f8417..3f160dd70c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -111,6 +111,26 @@ pcapkit-vendor = "pcapkit.vendor.__main__:main" cli = [ "emoji" ] # for ESP payload decryption, c.f. pcapkit.protocols.internet.esp crypto = [ "cryptography>=3.4" ] +# for NGAP decoding, c.f. pcapkit.protocols.application.ngap. Deliberately kept +# out of ``all``, for two reasons that are each sufficient on their own: +# +# * Size. ``pycrate`` ships every specification it has ever compiled in one +# distribution: installing it lands ~238 MB of ``pycrate_asn1dir`` (87 spec +# modules) to obtain the one 4.9 MB ``NGAP.py`` this needs. Paying that in +# ``all`` -- a 50x multiplier over the useful part -- to support one +# application protocol is not a trade to make on a user's behalf. +# * Licence. ``pycrate`` is LGPL-2.1+ where this package is BSD-3-Clause. That +# is fine as an optional import a user chooses, and it is not fine as +# something ``pip install pypcapkit[all]`` pulls in silently: LGPL carries +# obligations on redistribution that a BSD-licensed dependent may not want, +# and ``all`` reads as "the rest of the same thing" rather than as a licence +# decision. +# +# No version floor: NGAP has shipped precompiled in ``pycrate_asn1dir`` for many +# releases and the two entry points used here (``NGAP_PDU.from_aper`` and +# ``to_aper``) are long-standing. Verified against 0.8.1, which is pure Python +# with no compiled extension, so there is no toolchain prerequisite either. +NGAP = [ "pycrate" ] # for normal users DPKT = [ "dpkt" ] Scapy = [ "scapy" ] @@ -188,6 +208,9 @@ vendor = [ "requests[socks]", "beautifulsoup4[html5lib]" ] # all, but it and ``libpcap`` are published only as pre-releases, and ``all`` # should not be the way somebody ends up with a beta they did not ask for. It is # ``pip install pypcapkit[PCAP_CT]``, which says what it is installing. +# +# ``pycrate`` is left out for the size and licence reasons set out on the +# ``NGAP`` extra above; ``pip install pypcapkit[NGAP]`` is the way to it. all = [ "emoji", "cryptography>=3.4", diff --git a/tests/protocols/application/test_ngap_unit.py b/tests/protocols/application/test_ngap_unit.py new file mode 100644 index 0000000000..917f429132 --- /dev/null +++ b/tests/protocols/application/test_ngap_unit.py @@ -0,0 +1,487 @@ +# -*- coding: utf-8 -*- +"""Unit tests for :mod:`pcapkit.protocols.application.ngap`. + +The fixture is 58 bytes of real aligned PER, built inline rather than read from +a capture: this is a unit-tier module, so :file:`tests/conftest.py` will not let +it read a generated capture, and an SCTP frame carrying it is cheap to construct +with :meth:`SCTP.make `. + +Two axes have to keep working, and only one of them needs |pycrate|_: + +* **With** it, the fixture decodes and every surfaced field is asserted against + the value that was encoded, and the encoding round-trips byte for byte. +* **Without** it, :mod:`pcapkit` still imports, :class:`NGAP` is still + registered, and an NGAP payload degrades rather than raising. Those tests run + unconditionally -- the missing-dependency path is reached by resetting the + module's import cache, so it is exercised on a machine that *has* |pycrate|_ + too, which is the only way it gets covered in CI. + +.. |pycrate| replace:: ``pycrate`` +.. _pycrate: https://github.com/pycrate-org/pycrate + +""" + +from __future__ import annotations + +import importlib.util +import sys +import unittest +from unittest import mock + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +HAS_PYCRATE = importlib.util.find_spec('pycrate_asn1dir') is not None + +#: An ``NGSetupRequest`` in aligned PER, 58 octets. Four IEs: Global RAN Node ID +#: (a ``globalGNB-ID`` whose ``gNB-ID`` is the 24-bit string ``0x000102``), RAN +#: Node Name ``pcapkit-gnb``, a one-entry Supported TA List for TAC ``0x000001``, +#: and Default Paging DRX ``v128``. +NGSETUP_REQUEST = bytes.fromhex( + '00150036000004001b00080002f839100001020052400d0500706361706b69742d676e62' + '0066000d00000000010002f839000000080015400140' +) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class NGAPUnitTests(unittest.TestCase): + + ########################################################################## + # Helpers. + ########################################################################## + + @staticmethod + def _protocol(): + """An :class:`NGAP` instance with no packet bound to it.""" + from pcapkit.protocols.application.ngap import NGAP + + return object.__new__(NGAP) + + @staticmethod + def _sctp_frame(payload: bytes, ppid: int = 60) -> bytes: + """An SCTP packet whose single DATA chunk carries ``payload``.""" + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + proto = SCTP.__new__(SCTP) + return SCTP.make( + proto, srcport=9899, dstport=38412, vtag=0x11223344, + chunks=[(Chunk.Payload_Data, dict(I=True, U=True, B=True, E=True, tsn=1, + stream_id=0, stream_seq=0, ppid=ppid, + data=payload))], + ).pack() + + ########################################################################## + # Enumerations. These need no pycrate -- that is the point of them being + # pasted in rather than generated at import time. + ########################################################################## + + def test_enumerations_are_defined_without_pycrate(self) -> None: + from pcapkit.protocols.application.ngap import (Criticality, PDUKind, ProcedureCode, + ProtocolIE) + + # Counts as measured against pycrate 0.8.1's NGAP_Constants. A drift + # here means the pasted-in block was regenerated against a different + # specification revision, which is worth noticing deliberately. + self.assertEqual(len(ProcedureCode), 81) + self.assertEqual(len(ProtocolIE), 438) + self.assertEqual(len(Criticality), 3) + self.assertEqual(len(PDUKind), 3) + + self.assertEqual(int(ProcedureCode.NGSetup), 21) + self.assertEqual(int(ProcedureCode.InitialContextSetup), 14) + self.assertEqual(int(ProtocolIE.GlobalRANNodeID), 27) + self.assertEqual(int(ProtocolIE.RANNodeName), 82) + self.assertEqual(int(ProtocolIE.SupportedTAList), 102) + + # An IE's ID name and its open type's name differ here, which is why + # ``IE.type`` exists alongside ``IE.id``. + self.assertEqual(int(ProtocolIE.DefaultPagingDRX), 21) + + self.assertEqual(PDUKind.INITIATING_MESSAGE, 'initiatingMessage') + self.assertEqual(PDUKind.SUCCESSFUL_OUTCOME, 'successfulOutcome') + self.assertEqual(PDUKind.UNSUCCESSFUL_OUTCOME, 'unsuccessfulOutcome') + + def test_criticality_resolves_names_and_rejects_a_fourth_value(self) -> None: + from pcapkit.protocols.application.ngap import Criticality + + # pycrate hands back the ASN.1 identifier as a string, so the member + # names have to be spelled that way for the lookup to be direct. + self.assertIs(Criticality.get('reject'), Criticality.reject) + self.assertIs(Criticality.get('ignore'), Criticality.ignore) + self.assertIs(Criticality.get('notify'), Criticality.notify) + self.assertIs(Criticality.get(2), Criticality.notify) + self.assertIs(Criticality.get(Criticality.reject), Criticality.reject) + self.assertEqual(Criticality.reject.name, 'reject') + + # Closed ENUMERATED: three values and no extension marker, so a fourth + # is a bug rather than a newer release. + with self.assertRaises(ValueError): + Criticality(3) + + def test_unknown_procedure_and_ie_extend_rather_than_raise(self) -> None: + from pcapkit.protocols.application.ngap import ProcedureCode, ProtocolIE + + # A pycrate carrying a newer NGAP than the pasted block decodes IEs + # this package does not name; they must report, not explode. + self.assertEqual(ProcedureCode(200).name, 'Unassigned_200') + self.assertEqual(int(ProcedureCode(200)), 200) + self.assertEqual(ProtocolIE(113).name, 'Unassigned_113') # a real gap + self.assertEqual(ProtocolIE(9000).name, 'Unassigned_9000') + + # ProcedureCode is INTEGER (0..255) and ProtocolIE-ID is (0..65535). + with self.assertRaises(ValueError): + ProcedureCode(256) + with self.assertRaises(ValueError): + ProtocolIE(65536) + + ########################################################################## + # Registration and metadata. No pycrate needed. + ########################################################################## + + def test_registered_on_both_ngap_payload_protocol_identifiers(self) -> None: + from pcapkit.const.sctp.payload_protocol_identifier import PayloadProtocolIdentifier + from pcapkit.protocols.application.ngap import NGAP + from pcapkit.protocols.transport.sctp import SCTP + + for ppid in (PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NG_Application_Protocol, # noqa: E501 + PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NGAP_over_DTLS_over_SCTP): # noqa: E501 + with self.subTest(ppid=int(ppid)): + self.assertIn(ppid, SCTP.__proto__) + entry = SCTP.__proto__[ppid] + # A ModuleDescriptor until the first dispatch resolves it. + module = getattr(entry, 'module', None) + if module is None: + self.assertIs(entry, NGAP) + else: + self.assertEqual(module, 'pcapkit.protocols.application.ngap') + self.assertEqual(entry.name, 'NGAP') + + def test_name_length_and_length_hint(self) -> None: + from pcapkit.protocols.application.ngap import NGAP + + ngap = self._protocol() + self.assertEqual(ngap.name, 'NG Application Protocol') + self.assertEqual(ngap.layer, 'Application') + self.assertEqual(ngap.__length_hint__(), 4) + + # ``length`` is the whole PDU: NGAP prefixes it with no header. + ngap.__header__ = mock.Mock(data=NGSETUP_REQUEST) + self.assertEqual(ngap.length, 58) + + # An application protocol carries no numeral registry index. + from pcapkit.utilities.exceptions import IntError + with self.assertRaises(IntError): + NGAP.__index__() + + def test_exported_from_the_package_namespaces(self) -> None: + import pcapkit + from pcapkit.protocols.application import NGAP as ApplicationNGAP + from pcapkit.protocols.application.ngap import NGAP + from pcapkit.protocols.data.application import NGAP as DataNGAP + from pcapkit.protocols.schema.application import NGAP as SchemaNGAP + + self.assertIs(pcapkit.NGAP, NGAP) + self.assertIs(ApplicationNGAP, NGAP) + self.assertIn('NGAP', pcapkit.protocols.__proto__) + self.assertIs(pcapkit.protocols.__proto__['NGAP'], NGAP) + self.assertIsNot(DataNGAP, NGAP) + self.assertIsNot(SchemaNGAP, NGAP) + + ########################################################################## + # The missing-dependency path, exercised whether or not pycrate is present. + ########################################################################## + + def test_read_without_pycrate_raises_protocol_error(self) -> None: + from pcapkit.protocols.application import ngap as ngap_module + from pcapkit.utilities.exceptions import ProtocolError + + ngap = self._protocol() + ngap.__cached__ = {} + ngap._data = NGSETUP_REQUEST + ngap.__header__ = mock.Mock(data=NGSETUP_REQUEST) + + with mock.patch.object(ngap_module, '_PYCRATE', None): + with self.assertRaises(ProtocolError) as caught: + ngap.read() + self.assertIn('pycrate', str(caught.exception)) + self.assertIn('pypcapkit[NGAP]', str(caught.exception)) + + def test_make_without_pycrate_raises_protocol_error(self) -> None: + from pcapkit.protocols.application import ngap as ngap_module + from pcapkit.utilities.exceptions import ProtocolError + + ngap = self._protocol() + + with mock.patch.object(ngap_module, '_PYCRATE', None): + with self.assertRaises(ProtocolError) as caught: + ngap.make(procedure=21, message='NGSetupRequest', value={'protocolIEs': []}) + self.assertIn('pycrate', str(caught.exception)) + + def test_load_pycrate_caches_a_failed_import(self) -> None: + from pcapkit.protocols.application import ngap as ngap_module + + # ``None`` in sys.modules makes ``from pycrate_asn1dir.NGAP import ...`` + # raise ImportError, which is the branch a machine without pycrate takes. + with mock.patch.object(ngap_module, '_PYCRATE', NotImplemented): + with mock.patch.dict(sys.modules, {'pycrate_asn1dir.NGAP': None}): + self.assertIsNone(ngap_module.load_pycrate()) + # Cached, so a capture full of NGAP does not retry the import. + self.assertIsNone(ngap_module._PYCRATE) + self.assertIsNone(ngap_module.load_pycrate()) + + def test_make_requires_procedure_and_message(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + ngap = self._protocol() + for kwargs in ({}, {'procedure': 21}, {'message': 'NGSetupRequest'}): + with self.subTest(**kwargs): + with self.assertRaises(ProtocolError): + ngap.make(**kwargs) + + def test_make_from_raw_bytes_needs_no_pycrate(self) -> None: + from pcapkit.protocols.application import ngap as ngap_module + + ngap = self._protocol() + with mock.patch.object(ngap_module, '_PYCRATE', None): + schema = ngap.make(data=NGSETUP_REQUEST) + self.assertEqual(schema.data, NGSETUP_REQUEST) + + def test_sctp_payload_degrades_to_raw_without_pycrate(self) -> None: + """A capture parses end to end with no |pycrate|_ installed. + + This is the behaviour the optional dependency buys its optionality with: + the payload reaches :class:`~pcapkit.protocols.misc.raw.Raw` through + :func:`~pcapkit.utilities.decorators.beholder` rather than aborting the + frame, and keeps the PPID's name while doing so. + + """ + import io + + from pcapkit.protocols.application import ngap as ngap_module + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.sctp import SCTP + + raw = self._sctp_frame(NGSETUP_REQUEST) + with mock.patch.object(ngap_module, '_PYCRATE', None): + packet = SCTP(io.BytesIO(raw), len(raw)) + + self.assertEqual(int(packet.ppid), 60) + self.assertIsInstance(packet.payload, Raw) + self.assertEqual(bytes(packet.payload), NGSETUP_REQUEST) + self.assertEqual(str(packet.protochain), + 'SCTP:PayloadProtocolIdentifier_3GPP_NG_Application_Protocol') + + ########################################################################## + # Decoding. These need pycrate. + ########################################################################## + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_fixture_decodes_and_surfaces_the_promised_fields(self) -> None: + from pcapkit.protocols.application.ngap import (NGAP, Criticality, PDUKind, + ProcedureCode, ProtocolIE) + + info = NGAP(NGSETUP_REQUEST).info + + self.assertIs(info.kind, PDUKind.INITIATING_MESSAGE) + self.assertIs(info.procedure, ProcedureCode.NGSetup) + self.assertIs(info.criticality, Criticality.reject) + self.assertEqual(info.message, 'NGSetupRequest') + self.assertEqual(len(info.ies), 4) + + self.assertEqual( + [(ie.id, ie.criticality, ie.type) for ie in info.ies], + [(ProtocolIE.GlobalRANNodeID, Criticality.reject, 'GlobalRANNodeID'), + (ProtocolIE.RANNodeName, Criticality.ignore, 'RANNodeName'), + (ProtocolIE.SupportedTAList, Criticality.reject, 'SupportedTAList'), + # IE 21 is id-DefaultPagingDRX; its open type is PagingDRX. + (ProtocolIE.DefaultPagingDRX, Criticality.ignore, 'PagingDRX')], + ) + + # ``value`` holds the message body converted in full, so ``ies`` is a + # view of it rather than the only way in. + self.assertEqual(len(info.value['protocolIEs']), 4) + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_generic_conversion_preserves_asn1_shapes(self) -> None: + from pcapkit.protocols.application.ngap import NGAP + from pcapkit.protocols.data.application.ngap import BitString, Choice, Sequence + + info = NGAP(NGSETUP_REQUEST).info + + # CHOICE -> Choice, keeping the alternative's name. + node_id = info.ies[0].value + self.assertIsInstance(node_id, Choice) + self.assertEqual(node_id.name, 'globalGNB-ID') + self.assertIsInstance(node_id.value, Sequence) + + # OCTET STRING stays bytes; a hyphenated ASN.1 name is reachable by + # subscription, which is why the conversion does not rewrite keys. + self.assertEqual(node_id.value['pLMNIdentity'], b'\x02\xf8\x39') + gnb = node_id.value['gNB-ID'] + self.assertIsInstance(gnb, Choice) + + # BIT STRING -> BitString, so the bit length survives. + self.assertIsInstance(gnb.value, BitString) + self.assertEqual(gnb.value.value, 0x000102) + self.assertEqual(gnb.value.length, 24) + + # SEQUENCE OF -> list; PrintableString / ENUMERATED stay str. + self.assertEqual(info.ies[1].value, 'pcapkit-gnb') + self.assertEqual(info.ies[3].value, 'v128') + ta_list = info.ies[2].value + self.assertIsInstance(ta_list, list) + self.assertEqual(len(ta_list), 1) + self.assertEqual(ta_list[0]['tAC'], b'\x00\x00\x01') + slices = ta_list[0]['broadcastPLMNList'][0]['tAISliceSupportList'] + self.assertEqual(slices[0]['s-NSSAI']['sST'], b'\x01') + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_read_make_round_trip_is_byte_exact(self) -> None: + from pcapkit.protocols.application.ngap import NGAP + + ngap = NGAP(NGSETUP_REQUEST) + rebuilt = ngap.make(**NGAP._make_data(ngap.info)) + self.assertEqual(rebuilt.data, NGSETUP_REQUEST) + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_make_data_reports_every_field_make_consumes(self) -> None: + from pcapkit.protocols.application.ngap import NGAP + + info = NGAP(NGSETUP_REQUEST).info + kwargs = NGAP._make_data(info) + + self.assertEqual(set(kwargs), + {'kind', 'procedure', 'criticality', 'message', 'value', 'data'}) + self.assertIs(kwargs['kind'], info.kind) + self.assertIs(kwargs['procedure'], info.procedure) + self.assertIsNone(kwargs['data']) + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_make_accepts_plain_python_values(self) -> None: + from pcapkit.protocols.application.ngap import NGAP, Criticality, ProcedureCode + + ngap = self._protocol() + schema = ngap.make( + kind='initiatingMessage', + procedure=ProcedureCode.NGSetup, + criticality=Criticality.reject, + message='NGSetupRequest', + value={'protocolIEs': [ + {'id': 82, 'criticality': 'ignore', + 'value': ('RANNodeName', 'pcapkit-gnb')}, + ]}, + ) + # Decodes back to what went in, which is the check that matters -- the + # byte string itself is pycrate's business. + info = NGAP(schema.data).info + self.assertEqual(info.message, 'NGSetupRequest') + self.assertEqual(len(info.ies), 1) + self.assertEqual(info.ies[0].value, 'pcapkit-gnb') + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_make_rejects_a_message_the_specification_cannot_encode(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + ngap = self._protocol() + with self.assertRaises(ProtocolError) as caught: + ngap.make(procedure=21, message='NotAMessageType', value={}) + self.assertIn('cannot encode', str(caught.exception)) + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_malformed_payloads_raise_protocol_error_not_pycrate_errors(self) -> None: + """Nothing from |pycrate|_'s own exception hierarchies may escape. + + It raises from two unrelated ones -- ``pycrate_asn1rt.err.ASN1Err`` and + ``pycrate_core.charpy.CharpyErr`` -- and a truncated payload can also + surface as ``KeyError`` or ``IndexError`` from inside the codec. Every + one of those has to arrive as + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, or an NGAP packet + fails differently from every other protocol in the library. + + """ + from pcapkit.protocols.application.ngap import NGAP + from pcapkit.utilities.exceptions import ProtocolError + + cases = { + 'truncated header': NGSETUP_REQUEST[:2], + 'truncated body': NGSETUP_REQUEST[:7], + 'truncated mid-IE': NGSETUP_REQUEST[:40], + 'invalid choice index': b'\xff' * 8, + 'not asn.1 at all': bytes(range(16)), + 'ascii text': b'this is not a PDU', + 'trailing garbage': NGSETUP_REQUEST + b'\xde\xad\xbe\xef', + } + for label, payload in cases.items(): + with self.subTest(case=label): + try: + info = NGAP(payload).info + except ProtocolError as exc: + self.assertIn('NGAP:', str(exc)) + else: + # Decoding a prefix or ignoring a suffix is legitimate PER + # behaviour; what matters is that it did not raise something + # raw. Assert it produced a usable model. + self.assertIsNotNone(info.message) + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_end_to_end_through_sctp(self) -> None: + """PPID 60 dispatches an SCTP DATA chunk to :class:`NGAP`.""" + import io + + from pcapkit.protocols.application.ngap import NGAP, ProcedureCode + from pcapkit.protocols.transport.sctp import SCTP + + raw = self._sctp_frame(NGSETUP_REQUEST) + packet = SCTP(io.BytesIO(raw), len(raw)) + + self.assertEqual(int(packet.ppid), 60) + self.assertIsInstance(packet.payload, NGAP) + self.assertIs(packet.payload.info.procedure, ProcedureCode.NGSetup) + self.assertEqual(packet.payload.info.message, 'NGSetupRequest') + self.assertEqual(str(packet.protochain), 'SCTP:NGAP') + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_reset_val_is_never_called_on_the_parse_path(self) -> None: + """The ~105 ms trap, asserted rather than left to a comment. + + ``NGAP_PDU.reset_val()`` walks all 318 submodules of the compiled + specification. Calling it once per decode costs some 700x the decode + itself, and ``from_aper()`` overwrites the stored value regardless, so + it buys nothing. A future edit that adds it back is a 700x regression + with no visible symptom, which is exactly the kind of thing a test + should hold. + + """ + from pcapkit.protocols.application.ngap import NGAP, load_pycrate + + pdu = load_pycrate() + with mock.patch.object(type(pdu), 'reset_val') as reset: + NGAP(NGSETUP_REQUEST).info + reset.assert_not_called() + + @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') + def test_consecutive_decodes_do_not_share_state(self) -> None: + """The module-level PDU object is stateful; the models must not be. + + ``get_val()`` hands back the decoder's own containers rather than + copies, so a model that held a reference into them would be rewritten by + the next decode. Two decodes and a comparison is what catches that. + + """ + from pcapkit.protocols.application.ngap import NGAP + + first = NGAP(NGSETUP_REQUEST).info + snapshot = (first.message, [ie.id for ie in first.ies], + first.ies[1].value, first.ies[0].value.name) + + other = self._protocol().make( + procedure=9, message='ErrorIndication', value={'protocolIEs': []}) + second = NGAP(other.data).info + self.assertEqual(second.message, 'ErrorIndication') + + self.assertEqual((first.message, [ie.id for ie in first.ies], + first.ies[1].value, first.ies[0].value.name), snapshot) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_sctp_unit.py b/tests/protocols/transport/test_sctp_unit.py index 1ed07a51f1..6c031b774c 100644 --- a/tests/protocols/transport/test_sctp_unit.py +++ b/tests/protocols/transport/test_sctp_unit.py @@ -19,6 +19,7 @@ from __future__ import annotations +import contextlib import importlib.util import unittest from unittest import mock @@ -58,7 +59,8 @@ class SCTPUnitTests(unittest.TestCase): # NOTE: Unlike the sibling TCP/UDP unit tests, this module does not purge # and re-import :mod:`pcapkit` per test: nothing here depends on import-time # behaviour, and the re-import costs several seconds a test. Every test that - # mutates a class-level registry restores it in a ``finally``. + # mutates a class-level registry restores it -- ``__proto__`` through the + # ``_proto_registry`` helper below, the sub-registries in a ``finally``. ########################################################################## # Helpers. @@ -73,6 +75,28 @@ def _packet(raw: bytes): return SCTP(io.BytesIO(raw), len(raw)) + @staticmethod + @contextlib.contextmanager + def _proto_registry(): + """Restore ``SCTP.__proto__`` on the way out, defaults included. + + ``SCTP.__proto__.clear()`` was a sound teardown while the registry + started empty. It is not one now: NGAP is registered on PPIDs 60 and 66 + by default, and clearing the registry leaves every later test in the + process running against a registry that no import repopulates -- an + order-dependent failure that only appears in whichever test happens to + run second. + + """ + from pcapkit.protocols.transport.sctp import SCTP + + saved = dict(SCTP.__proto__) + try: + yield SCTP.__proto__ + finally: + SCTP.__proto__.clear() + SCTP.__proto__.update(saved) + @staticmethod def _build(chunks, **kwargs) -> bytes: """Construct a whole SCTP packet from a list of chunk specifications.""" @@ -832,23 +856,48 @@ def test_data_chunk_constructor_requires_user_data(self) -> None: ########################################################################## def test_ppid_dispatch_hook(self) -> None: + """The two NGAP PPIDs dispatch, and a junk payload still degrades. + + Both PPIDs are registered as defaults on + :attr:`SCTP.__proto__ ` + rather than by a ``register_sctp`` call at import time, so the assertion + is on the registry's declared contents. + + ``b'ngap-pdu'`` is not an aligned PER ``NGAP-PDU``, which is the point: + the failure has to reach :class:`Raw` through + :func:`~pcapkit.utilities.decorators.beholder` rather than escape, and + the payload has to keep the PPID's name while doing so. That path is + also what a capture parsed *without* ``pycrate`` installed takes on + every NGAP packet. + + """ from pcapkit.const.sctp.chunk import Chunk from pcapkit.const.sctp.payload_protocol_identifier import PayloadProtocolIdentifier + from pcapkit.protocols.application.ngap import NGAP from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport import sctp as sctp_module from pcapkit.protocols.transport.sctp import SCTP NGAP_PPID = PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NG_Application_Protocol # noqa: E501 + DTLS_PPID = PayloadProtocolIdentifier.PayloadProtocolIdentifier_3GPP_NGAP_over_DTLS_over_SCTP # noqa: E501 self.assertEqual(int(NGAP_PPID), 60) + self.assertEqual(int(DTLS_PPID), 66) + + for ppid in (NGAP_PPID, DTLS_PPID): + with self.subTest(ppid=int(ppid)): + self.assertIn(ppid, SCTP.__proto__) + entry = SCTP.__proto__[ppid] + module = getattr(entry, 'module', None) + if module is None: # already imported by an earlier test + self.assertIs(entry, NGAP) + else: + self.assertEqual(module, 'pcapkit.protocols.application.ngap') + self.assertEqual(entry.name, 'NGAP') raw = self._build([(Chunk.Payload_Data, dict(I=True, U=True, B=True, E=True, tsn=1, ppid=NGAP_PPID, data=b'ngap-pdu'))]) - # With nothing registered on the PPID, the payload is raw -- but it is - # still named after the PPID it arrived with, the way an unregistered - # transport type is (``IPv4:Use_for_experimentation_and_testing_253``), - # rather than being anonymised to a bare ``Raw``. - self.assertNotIn(NGAP_PPID, SCTP.__proto__) proto = self._packet(raw) self.assertEqual(proto.ppid, NGAP_PPID) self.assertIsInstance(proto.payload, Raw) @@ -856,17 +905,19 @@ def test_ppid_dispatch_hook(self) -> None: self.assertEqual(str(proto.protochain), 'SCTP:PayloadProtocolIdentifier_3GPP_NG_Application_Protocol') - # This is exactly the call a future NGAP class makes. - try: - SCTP.register(NGAP_PPID, Raw) + # A PPID registered over the default dispatches to the new class. The + # overwrite warning is expected here -- 60 is no longer a free slot -- + # and is asserted rather than allowed to litter the test output. + with self._proto_registry(): + with mock.patch.object(sctp_module, 'warn') as warned: + SCTP.register(NGAP_PPID, Raw) + self.assertEqual(warned.call_count, 1) self.assertIs(SCTP.__proto__[NGAP_PPID], Raw) self.assertIs(SCTP.__proto__[60], Raw) proto = self._packet(raw) self.assertIsInstance(proto.payload, Raw) self.assertEqual(bytes(proto.payload), b'ngap-pdu') - finally: - SCTP.__proto__.clear() def test_unregistered_ppid_does_not_mutate_the_class_registry(self) -> None: """An unregistered PPID reaches :class:`Raw` without touching ``__proto__``. @@ -899,7 +950,7 @@ def test_unregistered_ppid_does_not_mutate_the_class_registry(self) -> None: dict(I=True, U=True, B=True, E=True, tsn=1, ppid=UNREGISTERED, data=b'unknown-pdu'))]) - try: + with self._proto_registry(): before = set(SCTP.__proto__) self.assertNotIn(UNREGISTERED, before) @@ -934,8 +985,6 @@ def test_unregistered_ppid_does_not_mutate_the_class_registry(self) -> None: with mock.patch.object(sctp_module, 'warn') as warned: SCTP.register(UNREGISTERED, Raw) self.assertEqual(warned.call_count, 0) - finally: - SCTP.__proto__.clear() def test_packet_without_a_data_chunk_has_no_payload(self) -> None: from pcapkit.const.sctp.chunk import Chunk @@ -975,25 +1024,25 @@ def test_register_warns_on_overwrite(self) -> None: from pcapkit.protocols.transport import sctp as sctp_module from pcapkit.protocols.transport.sctp import SCTP - try: - SCTP.register(60, Raw) + # 4243 rather than 60: NGAP holds 60 by default, so registering it once + # would already be the overwrite this test means to trigger on the + # *second* call, and the assertion on the call count would pass for the + # wrong reason. + with self._proto_registry(): + SCTP.register(4243, Raw) with mock.patch.object(sctp_module, 'warn') as warned: - SCTP.register(60, Raw) + SCTP.register(4243, Raw) self.assertEqual(warned.call_count, 1) self.assertIn('payload protocol identifier', warned.call_args.args[0]) - finally: - SCTP.__proto__.clear() def test_register_sctp_wrapper_writes_the_ppid_registry(self) -> None: from pcapkit.foundation.registry.protocols import register_sctp from pcapkit.protocols.misc.raw import Raw from pcapkit.protocols.transport.sctp import SCTP - try: - register_sctp(60, 'pcapkit.protocols.misc.raw', 'Raw') - self.assertIs(SCTP.__proto__[60], Raw) - finally: - SCTP.__proto__.clear() + with self._proto_registry(): + register_sctp(4243, 'pcapkit.protocols.misc.raw', 'Raw') + self.assertIs(SCTP.__proto__[4243], Raw) ########################################################################## # Sub-registry registration. From 8efde450b17b77ebc70367f4e0ecf54cb99b2c9d Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 16 Sep 2026 13:40:44 -0400 Subject: [PATCH 3/6] docs: add the NGAP protocol page (#251) Follows the convention the sibling application pages use: the module docstring copied into the `.rst` under a `.. module::` directive, then `autoclass` per class, rather than `automodule` -- which was the review feedback on #408. `ProcedureCode` and `ProtocolIE` are rendered with `:no-members:`. Between them they carry 519 members, each named for the 3GPP identifier it came from, and a page listing all of them is longer than the specification's own tables and no more useful; a note says where to read them instead, and why there is no matching `pcapkit.vendor` module. No `pycrate` intersphinx mapping: the project publishes no `objects.inv`, so the link is a plain external reference through the same `|pycrate|_` substitution `esp.rst` uses for `cryptography`. One line added to the application-layer toctree. Sphinx build verified against a clean export of HEAD with the same interpreter: 285 warnings before, 285 after -- none added, none removed. --- .../pcapkit/protocols/application/index.rst | 1 + .../pcapkit/protocols/application/ngap.rst | 192 ++++++++++++++++++ 2 files changed, 193 insertions(+) create mode 100644 docs/source/pcapkit/protocols/application/ngap.rst diff --git a/docs/source/pcapkit/protocols/application/index.rst b/docs/source/pcapkit/protocols/application/index.rst index 27b93880c8..0423d668e4 100644 --- a/docs/source/pcapkit/protocols/application/index.rst +++ b/docs/source/pcapkit/protocols/application/index.rst @@ -16,6 +16,7 @@ application layer, with detailed implementation and methods. httpv1 httpv2 ftp + ngap .. todo:: diff --git a/docs/source/pcapkit/protocols/application/ngap.rst b/docs/source/pcapkit/protocols/application/ngap.rst new file mode 100644 index 0000000000..a12aeed9a0 --- /dev/null +++ b/docs/source/pcapkit/protocols/application/ngap.rst @@ -0,0 +1,192 @@ +NGAP - NG Application Protocol +============================== + +.. module:: pcapkit.protocols.application.ngap + +:mod:`pcapkit.protocols.application.ngap` contains +:class:`~pcapkit.protocols.application.ngap.NGAP` only, +which implements extractor for the NG Application Protocol +(NGAP) [*]_, as specified in 3GPP TS 38.413. + +NGAP is the control plane between a 5G RAN node (gNB or ng-eNB) and an AMF. +It runs over SCTP and is named by the DATA chunk's *payload protocol +identifier* rather than by a port, so :class:`NGAP` is registered on +:attr:`SCTP.__proto__ ` under +PPID 60 (``NG_Application_Protocol``) and PPID 66 +(``NGAP_over_DTLS_over_SCTP``), c.f. +:func:`pcapkit.foundation.registry.protocols.register_sctp`. + +An SCTP DATA chunk that names NGAP carries exactly one ``NGAP-PDU``, encoded +in **aligned PER** (ALIGNED PACKED ENCODING RULES, :abbr:`APER`). There is no +header to read and no framing to resolve: the whole payload is the encoding, +and none of its structure is visible until an ASN.1 decoder has run over it. + +Decoding therefore needs the optional |pycrate|_ dependency +(``pip install pypcapkit[NGAP]``). :mod:`pcapkit` imports and works without +it; an NGAP payload simply degrades to the opaque payload path, exactly as an +unregistered PPID would, because +:meth:`SCTP._import_next_layer ` +is wrapped in :func:`~pcapkit.utilities.decorators.beholder` and falls back to +:class:`~pcapkit.protocols.misc.raw.Raw`. + +Why |pycrate|_ rather than a PER codec of our own +------------------------------------------------- + +Two things make it the cheaper answer. |pycrate|_ **ships NGAP already +compiled**, at ``pycrate_asn1dir/NGAP.py``, so the 3GPP ASN.1 source does not +have to be vendored here and tracked across releases; and it is **pure +Python**, with no compiled extension to build on any platform. Decoding costs +0.15 ms per PDU, the same order as :mod:`pcapkit`'s own per-packet cost, so +the generic strategy below is not paying for the convenience. + +Generic conversion, not 51 hand-written procedures +-------------------------------------------------- + +The decoded value tree is mapped into :class:`~pcapkit.corekit.infoclass.Info` +objects **structurally**, by ASN.1 shape rather than by procedure: + +=============================== ========================================================== +ASN.1 / |pycrate|_ shape :mod:`pcapkit` model +=============================== ========================================================== +``SEQUENCE`` / ``SET`` (a dict) :class:`~pcapkit.protocols.data.application.ngap.Sequence` +``SEQUENCE OF`` (a list) :obj:`list` +``CHOICE`` / open type :class:`~pcapkit.protocols.data.application.ngap.Choice` +``BIT STRING`` :class:`~pcapkit.protocols.data.application.ngap.BitString` +``INTEGER``, ``OCTET STRING``, kept as :obj:`int`, :obj:`bytes`, :obj:`str` +``ENUMERATED``, ``BOOLEAN`` +=============================== ========================================================== + +That is a deliberate trade. Every one of the 81 elementary procedures and 438 +protocol IEs works on the day it is decoded, and a new 3GPP release needs no +code change here; what is given up is per-IE typing, so an IE's value is +reported in the specification's own shape rather than as a +:mod:`pcapkit`-specific model. The fields worth reading at a glance -- the PDU +kind, procedure code, criticality, message type name and the IE list -- are +surfaced as first-class fields on +:class:`~pcapkit.protocols.data.application.ngap.NGAP` regardless. + +Known limitations +----------------- + +* **PPID 66 payloads are not decoded.** ``NGAP_over_DTLS_over_SCTP`` wraps the + ``NGAP-PDU`` in a DTLS record, and :mod:`pcapkit` implements no DTLS, so the + bytes reaching :meth:`NGAP.read` are not an APER encoding. The PPID is + registered so that it is *named* rather than anonymous; the payload itself + degrades to :class:`~pcapkit.protocols.misc.raw.Raw`. +* **The specification version is |pycrate|_'s, not this package's.** The IE and + procedure enumerations were generated from ``NGAP_Constants`` of |pycrate|_ + 0.8.1 (Release-18-era: 81 procedure codes, 438 protocol IE IDs, highest 443). + A |pycrate|_ that carries a newer NGAP will decode IEs that + :class:`~pcapkit.protocols.application.ngap.ProcedureCode` and + :class:`~pcapkit.protocols.application.ngap.ProtocolIE` do not name; both + extend themselves at lookup time rather than failing, so such a value is + reported as ``Unassigned_``. +* **NGAP over a fragmented SCTP association is not reassembled.** A DATA chunk + is decoded on its own, so an ``NGAP-PDU`` split across chunks by SCTP + fragmentation fails to decode rather than being reassembled first. +* **Private IEs (``PrivateMessage``) carry no schema.** Their contents are + vendor defined, so the generic conversion reports whatever ASN.1 shape the + encoding declares and cannot name the fields. + +.. autoclass:: pcapkit.protocols.application.ngap.NGAP + :no-members: + :show-inheritance: + + .. autoproperty:: name + .. autoproperty:: length + + .. automethod:: read + .. automethod:: make + + .. automethod:: __length_hint__ + .. automethod:: _make_data + +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.protocols.application.ngap.load_pycrate + +.. autodata:: pcapkit.protocols.application.ngap._PYCRATE + +.. autodata:: pcapkit.protocols.application.ngap._PDU_LOCK + :no-value: + +.. autofunction:: pcapkit.protocols.application.ngap._convert + +.. autofunction:: pcapkit.protocols.application.ngap._revert + +Auxiliary Data +-------------- + +.. autoclass:: pcapkit.protocols.application.ngap.PDUKind + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.application.ngap.Criticality + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.application.ngap.ProcedureCode + :no-members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.application.ngap.ProtocolIE + :no-members: + :show-inheritance: + +.. note:: + + :class:`~pcapkit.protocols.application.ngap.ProcedureCode` and + :class:`~pcapkit.protocols.application.ngap.ProtocolIE` are rendered without + their members on purpose: between them they carry 519 of them, each named for + the 3GPP identifier it comes from, and a page listing all of them is longer + than the specification's own tables and no more useful. Read them from + ``pcapkit/protocols/application/ngap.py``, or from + ``pycrate_asn1dir.NGAP.NGAP_Constants``, which is where they were generated + from. + + Unlike the enumerations under :mod:`pcapkit.const`, these are not crawled + from an IANA registry -- 3GPP publishes them in the ASN.1 of TS 38.413 rather + than in a registry with a stable page -- so there is no matching module under + :mod:`pcapkit.vendor`. + +Header Schemas +-------------- + +.. module:: pcapkit.protocols.schema.application.ngap + +.. autoclass:: pcapkit.protocols.schema.application.ngap.NGAP + :members: + :show-inheritance: + +Data Models +----------- + +.. module:: pcapkit.protocols.data.application.ngap + +.. autoclass:: pcapkit.protocols.data.application.ngap.NGAP + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.application.ngap.IE + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.application.ngap.Choice + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.application.ngap.BitString + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.application.ngap.Sequence + :members: + :show-inheritance: + +.. |pycrate| replace:: ``pycrate`` +.. _pycrate: https://github.com/pycrate-org/pycrate + +.. rubric:: Footnotes + +.. [*] https://en.wikipedia.org/wiki/NG_Application_Protocol From 99406c1325a042d87d2fc3f7c464d335852e340f Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 16 Sep 2026 14:40:15 -0400 Subject: [PATCH 4/6] tests: update the Raw fallback assertion the beholder alias fix invalidates CI's Integration leg failed on all six interpreters with one real failure: `test_malformed_http_payload_falls_back_to_raw` asserted `tcp.payload.info.protocol is None`, and forwarding `alias` from `beholder` makes it 80. Neither the unit tier nor the local runs cover that tier, which is how it reached CI. The new value is the right one -- `Data_Raw.protocol` is "the original enumeration of this protocol", and 80 is what the payload arrived as -- so the assertion is updated rather than the fix reverted. The protochain still reads `Raw`, since a plain int has no name to render. The comment justifying the forward was too broad, though, and is now measured rather than asserted. It claimed an unregistered code keeps its name generally. True on SCTP (an unknown PPID gives `SCTP:Unassigned_4243` and `protocol=4243`, which is why registering NGAP would otherwise have made a failed parse *less* informative than leaving PPID 60 alone) and on IP, but false on TCP, where `Transport._decode_next_layer` resolves ports through `__proto__` and reaches Raw without an alias, so an unknown port already reports `None`. So the failure path is now uniform across layers while the unknown-code paths remain inconsistent with each other -- filed as #418. Full suite, the command CI's Integration leg runs: 806 passed, 17 skipped. --- pcapkit/utilities/decorators.py | 29 ++++++++++++------- .../application/test_http_runtime.py | 13 ++++++++- 2 files changed, 31 insertions(+), 11 deletions(-) diff --git a/pcapkit/utilities/decorators.py b/pcapkit/utilities/decorators.py index 7cb8bb68c8..a946bee77f 100644 --- a/pcapkit/utilities/decorators.py +++ b/pcapkit/utilities/decorators.py @@ -150,16 +150,25 @@ def behold(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_beholder': # SCTP payload protocol identifier, which NGAP now is. file_ = self._get_payload() - # NOTE: ``alias=proto`` matches what the success path passes, so a - # payload that failed to parse is still named after the protocol - # number it arrived with -- ``SCTP:PayloadProtocolIdentifier_3GPP_NG - # _Application_Protocol`` rather than a bare ``SCTP:Raw``. Without - # it, registering a protocol on a number made the output *less* - # informative than leaving the number unregistered, since an - # unregistered number reaches Raw through the success path and keeps - # its name. A plain integer has no ``name`` and still renders as - # ``Raw``, c.f. ``Raw.__post_init__``, so this only adds a name where - # the registry key is an enumeration. + # NOTE: ``alias=proto`` matches what ``_import_next_layer`` passes, so + # a payload that failed to parse still reports the code it arrived + # with, which is what ``Data_Raw.protocol`` means. A plain integer has + # no ``name`` and still renders as ``Raw`` in the protochain, c.f. + # ``Raw.__post_init__``, so this only adds a name where the registry + # key is an enumeration. + # + # Measured, because the layers differ and it is easy to state this too + # broadly: SCTP's unregistered path keeps its enumeration -- an unknown + # PPID gives ``SCTP:Unassigned_4243`` and ``protocol=4243`` -- so + # without this line, *registering* NGAP on PPID 60 would have made a + # failed parse report a bare ``SCTP:Raw`` and ``protocol=None``, less + # than the same bytes gave while unregistered. TCP's unregistered path + # does not: an unknown port yields ``protocol=None`` already, because + # ``Transport._decode_next_layer`` resolves ports through + # ``__proto__`` and never reaches here. So this makes the *failure* + # path uniform while the *unknown* paths stay inconsistent with each + # other, which is #418 rather than something to fix from inside a + # decorator. next_ = protocol(file_, length, error=str(exc), alias=proto) return cast('R_beholder', next_) return behold diff --git a/tests/protocols/application/test_http_runtime.py b/tests/protocols/application/test_http_runtime.py index 1a6a0683da..8d9dee6df9 100644 --- a/tests/protocols/application/test_http_runtime.py +++ b/tests/protocols/application/test_http_runtime.py @@ -102,7 +102,18 @@ def test_malformed_http_payload_falls_back_to_raw(self) -> None: self.assertEqual(tcp.name, 'Transmission Control Protocol') self.assertEqual(type(tcp.payload).__name__, 'Raw') self.assertEqual(tcp.payload.name, 'Unknown') - self.assertIsNone(tcp.payload.info.protocol) + + # The registry key that selected the protocol whose parse then failed -- + # TCP port 80 here. It used to be :data:`None`, because ``beholder`` was + # the one path to :class:`Raw` that did not forward ``alias``; that made a + # *registered* number less informative than an unregistered one, which + # reaches Raw through ``_import_next_layer`` and keeps its enumeration. + # ``Data_Raw.protocol`` is "the original enumeration of this protocol", and + # 80 is what this payload arrived as, so the value belongs here. The + # protochain still reads ``Raw``: a plain :obj:`int` has no ``name`` to + # render, and only an enumeration key -- SCTP's payload protocol + # identifier, say -- shows through. + self.assertEqual(tcp.payload.info.protocol, 80) self.assertIsNotNone(tcp.payload.info.packet) From c3e8dce4be6bd1b8dff51d62d42f4f30466d8619 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 16 Sep 2026 14:54:55 -0400 Subject: [PATCH 5/6] ngap: correct the procedure count, and say that PPID 66 does not decode Four review findings on #417, all of them right. The "not 51 hand-written procedures" heading in `ngap.py` and `ngap.rst` was wrong: `ProcedureCode` has 81 members, and 51 was a bad number from the brief that the enums themselves already corrected. Now 81 in both, which is the same character count, so the RST underline is unaffected. The other two are accurate-but-misleading-by-omission. `SCTP`'s class docstring said both PPIDs are registered to NGAP without saying that only 60 can decode: a PPID 66 payload is an NGAP PDU inside a DTLS record, and with no DTLS implementation those bytes are never aligned PER, so 66 degrades to `Raw` on every capture rather than only when `pycrate` is absent. The module docstring and the docs page already said so; the class docstring, which is where someone reading `SCTP.__proto__` looks, did not. `test_ppid_dispatch_hook`'s docstring had the same gap, and its assertion is deliberately no stronger than "both dispatch" -- now stated, so the test is not read as claiming both decode. 56 tests pass across the SCTP and NGAP files. --- docs/source/pcapkit/protocols/application/ngap.rst | 2 +- pcapkit/protocols/application/ngap.py | 2 +- pcapkit/protocols/transport/sctp.py | 8 ++++++++ tests/protocols/transport/test_sctp_unit.py | 14 ++++++++++---- 4 files changed, 20 insertions(+), 6 deletions(-) diff --git a/docs/source/pcapkit/protocols/application/ngap.rst b/docs/source/pcapkit/protocols/application/ngap.rst index a12aeed9a0..c943f43585 100644 --- a/docs/source/pcapkit/protocols/application/ngap.rst +++ b/docs/source/pcapkit/protocols/application/ngap.rst @@ -39,7 +39,7 @@ Python**, with no compiled extension to build on any platform. Decoding costs 0.15 ms per PDU, the same order as :mod:`pcapkit`'s own per-packet cost, so the generic strategy below is not paying for the convenience. -Generic conversion, not 51 hand-written procedures +Generic conversion, not 81 hand-written procedures -------------------------------------------------- The decoded value tree is mapped into :class:`~pcapkit.corekit.infoclass.Info` diff --git a/pcapkit/protocols/application/ngap.py b/pcapkit/protocols/application/ngap.py index 589946fd04..ec552dce83 100644 --- a/pcapkit/protocols/application/ngap.py +++ b/pcapkit/protocols/application/ngap.py @@ -41,7 +41,7 @@ 0.15 ms per PDU, the same order as :mod:`pcapkit`'s own per-packet cost, so the generic strategy below is not paying for the convenience. -Generic conversion, not 51 hand-written procedures +Generic conversion, not 81 hand-written procedures -------------------------------------------------- The decoded value tree is mapped into :class:`~pcapkit.corekit.infoclass.Info` diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index b78ecba5f9..84ae8a9f3d 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -225,6 +225,14 @@ class SCTP(Transport[Data_SCTP, Schema_SCTP], (``NG_Application_Protocol``) and 66 (``NGAP_over_DTLS_over_SCTP``). Every other PPID resolves to :class:`~pcapkit.protocols.misc.raw.Raw`. + Only PPID 60 actually decodes, though. A PPID 66 payload is an NGAP PDU + wrapped in a DTLS record, and :mod:`pcapkit` has no DTLS implementation, so + those bytes are not aligned PER and the parse degrades to + :class:`~pcapkit.protocols.misc.raw.Raw` -- every time, not only when + ``pycrate`` is absent. It is registered so that the PPID is *named* in the + protochain rather than reported as an unassigned number, which is strictly + more than leaving it out would give. + This class currently supports parsing of the following SCTP chunks, which are directly mapped to the :class:`pcapkit.const.sctp.chunk.Chunk` enumeration: diff --git a/tests/protocols/transport/test_sctp_unit.py b/tests/protocols/transport/test_sctp_unit.py index 6c031b774c..cc73ce801e 100644 --- a/tests/protocols/transport/test_sctp_unit.py +++ b/tests/protocols/transport/test_sctp_unit.py @@ -856,19 +856,25 @@ def test_data_chunk_constructor_requires_user_data(self) -> None: ########################################################################## def test_ppid_dispatch_hook(self) -> None: - """The two NGAP PPIDs dispatch, and a junk payload still degrades. + """Both NGAP PPIDs dispatch, and a junk payload still degrades. Both PPIDs are registered as defaults on :attr:`SCTP.__proto__ ` rather than by a ``register_sctp`` call at import time, so the assertion is on the registry's declared contents. + Dispatching is all the two have in common, and the assertion is + deliberately no stronger than that. Only PPID 60 can decode: a PPID 66 + payload is an NGAP PDU inside a DTLS record, and with no DTLS + implementation those bytes are never aligned PER, so 66 degrades to + :class:`Raw` on every well-formed capture as well as on this junk one. + ``b'ngap-pdu'`` is not an aligned PER ``NGAP-PDU``, which is the point: the failure has to reach :class:`Raw` through :func:`~pcapkit.utilities.decorators.beholder` rather than escape, and - the payload has to keep the PPID's name while doing so. That path is - also what a capture parsed *without* ``pycrate`` installed takes on - every NGAP packet. + the payload has to keep the PPID's name while doing so. For PPID 60 that + path is also what a capture parsed *without* ``pycrate`` installed takes + on every NGAP packet. """ from pcapkit.const.sctp.chunk import Chunk From 07c355b1cae90c4f7847154a64aa98b8fd241393 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 16 Sep 2026 16:40:29 -0400 Subject: [PATCH 6/6] ngap: drop Criticality.get's unused default, and report one error type Copilot is right on #417: `Criticality.get` declared and documented a `default` it never used. An unknown name raised `KeyError`, an unknown value `ValueError` via `_missing_`, and `default` reached neither path. The behaviour is intended -- `Criticality` is an ASN.1 `ENUMERATED` with no extension marker, so a fourth value is unencodable and a lookup for one is a bug, which is why `_missing_` raises deliberately. It is the parameter that was wrong, not the closedness, so the parameter is gone. No caller passed it: all five call sites use one argument. Its siblings keep theirs, and the docstring now says why rather than leaving the difference to be rediscovered -- `ProcedureCode` and `ProtocolIE` extend through `extend_enum` for a code a newer specification names, which is right for a registry that grows and impossible for one that cannot. Also made the unknown-name path raise `ValueError` like the unknown-value path, so the two ways of getting this wrong no longer report differently. Verified: signature carries no `default`, valid int/str/member lookups unchanged, both unknown forms now `ValueError`. 22 NGAP tests and the SCTP file pass. --- pcapkit/protocols/application/ngap.py | 26 +++++++++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/pcapkit/protocols/application/ngap.py b/pcapkit/protocols/application/ngap.py index ec552dce83..8bf08ee586 100644 --- a/pcapkit/protocols/application/ngap.py +++ b/pcapkit/protocols/application/ngap.py @@ -212,12 +212,29 @@ class Criticality(IntEnum): notify = 2 @staticmethod - def get(key: 'int | str | Criticality', default: 'int' = -1) -> 'Criticality': + def get(key: 'int | str | Criticality') -> 'Criticality': """Backport support for original codes. + Unlike :meth:`ProcedureCode.get` and :meth:`ProtocolIE.get`, this takes + no ``default``: those two extend themselves through + :func:`~aenum.extend_enum` when a newer specification names a code this + release does not, which is the right answer for a registry that grows. + ``Criticality`` cannot grow. It is an ASN.1 ``ENUMERATED`` with no + extension marker, so a fourth value is unencodable and a lookup for one + is a bug rather than a version skew -- see :meth:`_missing_`. A + ``default`` parameter here would have to be ignored, and one that is + declared, documented and ignored is worse than one that is absent. + Args: key: Key to get enum item. - default: Default value if not found. + + Returns: + The matching member. + + Raises: + ValueError: If ``key`` names no member. Raised for an unknown name as + well as an unknown value, so that the two ways of getting this + wrong do not report differently. :meta private: """ @@ -225,7 +242,10 @@ def get(key: 'int | str | Criticality', default: 'int' = -1) -> 'Criticality': return key if isinstance(key, int): return Criticality(key) - return Criticality[key] # type: ignore[misc] + try: + return Criticality[key] # type: ignore[misc] + except KeyError: + raise ValueError('%r is not a valid %s' % (key, Criticality.__name__)) from None @classmethod def _missing_(cls, value: 'int') -> 'NoReturn':