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..c943f43585 --- /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 81 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 diff --git a/pcapkit/__init__.py b/pcapkit/__init__.py index e1fed082ae..8eba3c6c11 100644 --- a/pcapkit/__init__.py +++ b/pcapkit/__init__.py @@ -118,7 +118,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 8f57831f12..3db61aa816 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -116,7 +116,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 b457493e45..2e440fbb1b 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, ~typing.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..8bf08ee586 --- /dev/null +++ b/pcapkit/protocols/application/ngap.py @@ -0,0 +1,1144 @@ +# -*- 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 81 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') -> '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. + + 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: + """ + if isinstance(key, Criticality): + return key + if isinstance(key, int): + return Criticality(key) + 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': + """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 5b1cba3848..84ae8a9f3d 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -220,7 +220,18 @@ 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`. + + 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` @@ -343,7 +354,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/pcapkit/utilities/decorators.py b/pcapkit/utilities/decorators.py index f170fb732b..a946bee77f 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,38 @@ 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 ``_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/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_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) 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..cc73ce801e 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,54 @@ def test_data_chunk_constructor_requires_user_data(self) -> None: ########################################################################## def test_ppid_dispatch_hook(self) -> None: + """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. 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 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 +911,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 +956,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 +991,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 +1030,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. 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)