# -*- 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`,
which implements extractor for the NG Application Protocol
(NGAP) [*]_, as specified in 3GPP TS 38.413, and the enumerations its
decoded fields resolve to.
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__ <pcapkit.protocols.transport.sctp.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:`~pcapkit.protocols.protocol.Protocol._import_next_layer`, which
:class:`~pcapkit.protocols.transport.sctp.SCTP` inherits, is wrapped in
:func:`~pcapkit.corekit.decorators.beholder` and falls back to
:class:`~pcapkit.protocols.misc.raw.Raw`.
Decoder Dependency
------------------
|pycrate|_ is used rather than a PER codec of our own. It
**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
------------------
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``, ``ENUMERATED``, ``BOOLEAN`` kept as :obj:`int`, :obj:`bytes`, :obj:`str`, :obj:`bool`
========================================================== ==========================================================
That is a deliberate trade. Every elementary procedure and protocol IE
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 that of the installed decoder, not this
package's.** :class:`~pcapkit.const.ngap.procedure_code.ProcedureCode` and
:class:`~pcapkit.const.ngap.protocol_ie.ProtocolIE` were generated from
``NGAP_Constants`` of |pycrate|_ 0.8.1 (Release-18-era). A |pycrate|_ that
carries a newer
NGAP will decode values these enumerations do not name. Such a value
resolves to a non-registering member called ``Unassigned`` rather than
failing or adding a member to either enumeration -- see
:meth:`pcapkit.corekit.enum.EnumRegistry._unregistered_member`.
* **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
from pcapkit.const.ngap.procedure_code import ProcedureCode as Enum_ProcedureCode
from pcapkit.const.ngap.protocol_ie import ProtocolIE as Enum_ProtocolIE
from pcapkit.corekit.enum import EnumLookup
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']
#: Cached ``NGAP-PDU`` object, c.f. :func:`load_pycrate`. :data:`NotImplemented`
#: means the import has not been attempted yet, :data:`None` that it was
#: attempted and |pycrate|_ is not installed, and anything else is the loaded
#: object. "Not installed" is kept apart from "not attempted" so that 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()
[docs]
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.
#
# PDUKind and Criticality are closed ASN.1 ENUMERATED types with no extension
# marker, so a lookup for a value neither declares is a bug, not a version
# skew -- they stay hand-written here rather than generated, like the closed
# mh.py helper enums (#877).
#
# ProcedureCode and ProtocolIE are the other shape: 3GPP TS 38.413 keeps
# assigning new elementary procedures and IEs to both, but neither is an IANA
# registry with a crawlable page -- it is a PDF, with no CSV or HTML kept in
# step with it. They are sourced from |pycrate|_'s own compiled specification
# instead (pycrate_asn1dir.NGAP's NGAP_Constants, #880), the same optional
# dependency this module needs to decode NGAP at all. They therefore live under
# pcapkit.const.ngap, generated by pcapkit.vendor.ngap.procedure_code and
# pcapkit.vendor.ngap.protocol_ie, and are imported here as Enum_ProcedureCode
# and Enum_ProtocolIE; this module does not re-export them under their
# original names (#1744). Only the vendor crawler that generates the const module
# needs pycrate; neither this module nor the generated one does at import.
#
# The SCTP payload protocol identifiers *are* IANA's, and are used from
# pcapkit.const.sctp.payload_protocol_identifier rather than duplicated here.
##############################################################################
[docs]
class PDUKind(EnumLookup, 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.
Built on :class:`~pcapkit.corekit.enum.EnumLookup`, the lookup contract
shared by every non-registry enumeration (:issue:`877`). The class defines
neither ``get`` nor ``_missing_`` of its own.
"""
#: 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'
[docs]
class Criticality(EnumLookup, 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:`~pcapkit.corekit.enum.EnumLookup.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.
Carries no ``get`` of its own: the inherited
:meth:`~pcapkit.corekit.enum.EnumLookup.get` is used as is, so an unknown
*name* is :exc:`KeyError`-shaped, exactly as ``E['nosuch']`` is on a stdlib
:class:`~enum.Enum` (:issue:`923`), and an unknown *value* raises
:exc:`ValueError`.
A ``default`` is honoured only when it names a registered value:
``get('nosuch', Criticality.reject)`` answers ``reject``, and
``get('nosuch', 99)`` raises. A member-valued ``default`` resolves through
``_value2member_map_`` and never through the constructor, so on this closed
ASN.1 ``ENUMERATED`` with no extension marker no path can mint a fourth
value; see :meth:`_missing_`, which refuses one outright.
"""
#: Reject the whole message.
reject = 0
#: Ignore the IE and carry on.
ignore = 1
#: Ignore the IE, carry on, and report it.
notify = 2
[docs]
@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__))
##############################################################################
# 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 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.
##############################################################################
[docs]
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__ <NGAP.__length_hint__>`, which reports the
four-octet prefix common to every PDU rather than this PDU's size.
"""
return len(self.__header__.data)
##########################################################################
# Methods.
##########################################################################
[docs]
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, if the payload is
not a well formed aligned PER ``NGAP-PDU``, or if octets follow
the ``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 every
# submodule 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.
from pycrate_core.charpy import Charpy # pylint: disable=import-outside-toplevel
# NOTE: decoded from a `Charpy` rather than from the bytes, so
# that what the decoder did not consume can be measured: a bare
# `from_aper(bytes)` ignores a suffix. SCTP preserves message
# boundaries: each user message occupies its own DATA chunk
# (RFC 9260 §1.3), and one with both B and E set is a complete,
# unfragmented message (§3.3.1). The outermost APER encoding is
# padded with zero to seven zero bits up to the next octet
# (X.691 (02/2021) §11.1.4), and the result is the complete
# encoding (§11.1.5). So a whole octet left over is not part of
# the NGAP-PDU.
char = Charpy(schema.data)
pdu.from_aper(char)
trailing = char.len_byte()
if trailing:
raise ProtocolError(f'NGAP: {trailing} octet(s) after the NGAP-PDU')
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=Enum_ProtocolIE.get(ie['id']),
criticality=Criticality.get(ie['criticality']),
type=choice.name,
value=choice.value,
))
ngap = Data_NGAP(
kind=PDUKind(kind),
procedure=Enum_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, such as the trailing-octet
# check above, so it must not be caught below and relabelled
# "malformed NGAP-PDU".
raise
except Exception as exc:
# NOTE: `Exception` rather than something narrower on purpose.
# pycrate signals a bad encoding from 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
[docs]
def make(self,
kind: 'PDUKind | str' = PDUKind.INITIATING_MESSAGE,
procedure: 'Optional[Enum_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(Enum_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.
##########################################################################
[docs]
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.
##########################################################################
[docs]
@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,
}