revised IPv6-Opts schema based on changes in Schema and fields

Author: JarryShaw

pcapkit/foundation/registry/protocols.py

@@ -27,7 +27,7 @@
 from pcapkit.protocols.schema.internet.hip import Parameter as Schema_HIP_Parameter
 from pcapkit.protocols.schema.internet.hopopt import Option as Schema_HOPOPT_Option
 from pcapkit.protocols.schema.internet.ipv4 import Option as Schema_IPv4_Option
-from pcapkit.protocols.schema.internet.ipv6_opts import IPv6_Opts as Schema_IPv6_Opts
+from pcapkit.protocols.schema.internet.ipv6_opts import Option as Schema_IPv6_Opts_Option
 from pcapkit.protocols.schema.internet.ipv6_route import MAP_IPV6_ROUTE_DATA
 from pcapkit.protocols.schema.internet.mh import MAP_MH_DATA
 from pcapkit.protocols.schema.internet.mh import CGAParameter as Schema_MH_CGAParameter
@@ -94,7 +94,6 @@
     from pcapkit.protocols.misc.pcapng import RecordParser as PCAPNG_RecordParser
     from pcapkit.protocols.misc.pcapng import SecretsConstructor as PCAPNG_SecretsConstructor
     from pcapkit.protocols.misc.pcapng import SecretsParser as PCAPNG_SecretsParser
-    from pcapkit.protocols.schema.internet.ipv6_opts import Option as Schema_IPv6_Opts_Option
     from pcapkit.protocols.schema.internet.ipv6_route import \
         RoutingType as Schema_IPv6_Route_RoutingType
     from pcapkit.protocols.schema.internet.mh import CGAExtension as Schema_MH_CGAExtension
@@ -408,7 +407,7 @@ def register_ipv6_opts_option(code: 'IPv6_Option', meth: 'str | tuple[IPv6_Opts_
 
     IPv6_Opts.register_option(code, meth)
     if schema is not None:
-        cast('OptionField[Schema_IPv6_Opts_Option]', Schema_IPv6_Opts.options).registry[code] = schema
+        Schema_IPv6_Opts_Option.register(code, schema)
     logger.info('registered IPv6-Opts option parser: %s', code.name)
 
 

pcapkit/protocols/schema/internet/ipv6_opts.py

@@ -20,7 +20,7 @@
 from pcapkit.corekit.fields.numbers import (EnumField, NumberField, UInt8Field, UInt16Field,
                                             UInt32Field)
 from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField
-from pcapkit.protocols.schema.schema import Schema, schema_final
+from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final
 from pcapkit.utilities.exceptions import FieldValueError
 from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING
 
@@ -38,7 +38,7 @@
 
 if TYPE_CHECKING:
     from ipaddress import IPv4Address, IPv6Address
-    from typing import Any, Optional
+    from typing import Any, DefaultDict, Optional, Type
 
     from pcapkit.corekit.fields.field import _Field as Field
     from pcapkit.protocols.protocol import Protocol
@@ -149,19 +149,18 @@ def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field':
 
     Returns:
         * If ``mode`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
-          wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFIdentificationBasedDPDOption`
+          wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.SMFIdentificationBasedDPDOption`
           instance.
         * If ``mode`` is ``1``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
-          wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFHashBasedDPDOption`
+          wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.SMFHashBasedDPDOption`
           instance.
 
     """
     mode = Enum_SMFDPDMode.get(pkt['test']['mode'])
-    if mode == Enum_SMFDPDMode.I_DPD:
-        return SchemaField(length=pkt['test']['len'], schema=SMFIdentificationBasedDPDOption)
-    if mode == Enum_SMFDPDMode.H_DPD:
-        return SchemaField(length=pkt['test']['len'], schema=SMFHashBasedDPDOption)
-    raise FieldValueError(f'IPv6-Opts: invalid SMF DPD mode: {mode}')
+    schema = SMFDPDOption.registry[mode]
+    if schema is None:
+        raise FieldValueError(f'IPv6-Opts: invalid SMF DPD mode: {mode}')
+    return SchemaField(length=pkt['test']['len'], schema=schema)
 
 
 def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field':
@@ -209,26 +208,27 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field':
 
     Returns:
         * If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
-          wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartRequestOption`
+          wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.QuickStartRequestOption`
           instance.
         * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
-          wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartReportOption`
+          wrapped :class:`~pcapkit.protocols.schema.internet.ipv6_opts.QuickStartReportOption`
           instance.
 
     """
     func = Enum_QSFunction.get(pkt['flags']['func'])
     pkt['flags']['func'] = func
 
-    if func == Enum_QSFunction.Quick_Start_Request:
-        return SchemaField(length=5, schema=QuickStartRequestOption)
-    if func == Enum_QSFunction.Report_of_Approved_Rate:
-        return SchemaField(length=5, schema=QuickStartReportOption)
-    raise FieldValueError(f'IPv6-Opts: invalid QS function: {func}')
+    schema = QuickStartOption.registry[func]
+    if schema is None:
+        raise FieldValueError(f'IPv6-Opts: invalid QS function: {func}')
+    return SchemaField(length=5, schema=schema)
 
 
-class Option(Schema):
+class Option(EnumSchema[Enum_Option]):
     """Header schema for IPv6-Opts options."""
 
+    __default__ = lambda: UnassignedOption
+
     #: Option type.
     type: 'Enum_Option' = EnumField(length=1, namespace=Enum_Option)
     #: Option length (conditional in case of ``Pad1`` option).
@@ -265,7 +265,8 @@ def __init__(self, type: 'Enum_Option', len: 'int', data: 'bytes') -> 'None': ..
 
 
 @schema_final
-class PadOption(Option):
+class PadOption(Option, code=[Enum_Option.Pad1,
+                              Enum_Option.PadN]):
     """Header schema for IPv6-Opts padding options."""
 
     #: Padding.
@@ -276,7 +277,7 @@ def __init__(self, type: 'Enum_Option', len: 'int') -> 'None': ...
 
 
 @schema_final
-class TunnelEncapsulationLimitOption(Option):
+class TunnelEncapsulationLimitOption(Option, code=Enum_Option.Tunnel_Encapsulation_Limit):
     """Header schema for IPv6-Opts tunnel encapsulation limit options."""
 
     #: Tunnel encapsulation limit.
@@ -287,7 +288,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', limit: 'int') -> 'None': ...
 
 
 @schema_final
-class RouterAlertOption(Option):
+class RouterAlertOption(Option, code=Enum_Option.Router_Alert):
     """Header schema for IPv6-Opts router alert options."""
 
     #: Router alert.
@@ -298,7 +299,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', alert: 'Enum_RouterAlert') -
 
 
 @schema_final
-class CALIPSOOption(Option):
+class CALIPSOOption(Option, code=Enum_Option.CALIPSO):
     """Header schema for IPv6-Opts common architecture label IPv6 security options."""
 
     #: CALIPSO domain of interpretation.
@@ -350,16 +351,22 @@ def post_process(self, packet: 'dict[str, Any]') -> 'SMFDPDOption':
         return ret
 
 
-class SMFDPDOption(Option):
+# register ``_SMFDPDOption`` as ``SMF_DPD`` option
+Option.register(Enum_Option.SMF_DPD, _SMFDPDOption)
+
+
+class SMFDPDOption(Option, EnumSchema[Enum_SMFDPDMode]):
     """Header schema for IPv6-Opts simplified multicast forwarding duplicate packet
     detection (``SMF_DPD``) options."""
 
+    __enum__: 'DefaultDict[Enum_SMFDPDMode, Type[SMFDPDOption]]' = collections.defaultdict(lambda: None)  # type: ignore[arg-type,return-value]
+
     if TYPE_CHECKING:
         mode: 'Enum_SMFDPDMode'
 
 
 @schema_final
-class SMFIdentificationBasedDPDOption(SMFDPDOption):
+class SMFIdentificationBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.I_DPD):
     """Header schema for IPv6-Opts SMF identification-based DPD options."""
 
     test: 'SMFDPDTestFlag' = ForwardMatchField(BitField(length=1, namespace={
@@ -401,7 +408,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', info: 'TaggerIDInfo',
 
 
 @schema_final
-class SMFHashBasedDPDOption(SMFDPDOption):
+class SMFHashBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.H_DPD):
     """Header schema for IPv6-Opts SMF hash-based DPD options."""
 
     #: Hash assist value (HAV).
@@ -426,7 +433,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', hav: 'bytes') -> 'None': ...
 
 
 @schema_final
-class PDMOption(Option):
+class PDMOption(Option, code=Enum_Option.PDM):
     """Header schema for IPv6-Opts performance and diagnostic metrics (PDM) options."""
 
     #: Scale delta time last received (DTLR).
@@ -475,7 +482,11 @@ def post_process(self, packet: 'dict[str, Any]') -> 'QuickStartOption':
         return ret
 
 
-class QuickStartOption(Option):
+# register ``_QuickStartOption`` as ``Quick_Start`` option
+Option.register(Enum_Option.Quick_Start, _QuickStartOption)
+
+
+class QuickStartOption(Option, EnumSchema[Enum_QSFunction]):
     """Header schema for IPv6-Opts quick start options."""
 
     #: Flags.
@@ -489,7 +500,7 @@ class QuickStartOption(Option):
 
 
 @schema_final
-class QuickStartRequestOption(QuickStartOption):
+class QuickStartRequestOption(QuickStartOption, code=Enum_QSFunction.Quick_Start_Request):
     """Header schema for IPv6-Opts quick start request options."""
 
     #: QS time-to-live (TTL).
@@ -505,7 +516,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
 
 
 @schema_final
-class QuickStartReportOption(QuickStartOption):
+class QuickStartReportOption(QuickStartOption, code=Enum_QSFunction.Report_of_Approved_Rate):
     """Header schema for IPv6-Opts quick start report of approved rate options."""
 
     #: Reserved.
@@ -521,7 +532,8 @@ def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
 
 
 @schema_final
-class RPLOption(Option):
+class RPLOption(Option, code=[Enum_Option.RPL_Option_0x23,
+                              Enum_Option.RPL_Option_0x63]):
     """Header schema for IPv6-Opts routing protocol for low-power and lossy networks (RPL) options."""
 
     #: Flags.
@@ -541,7 +553,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', flags: 'RPLFlags', id: 'int'
 
 
 @schema_final
-class MPLOption(Option):
+class MPLOption(Option, code=Enum_Option.MPL_Option):
     """Header schema for IPv6-Opts multicast protocol for low-power and lossy networks (MPL) options."""
 
     #: Flags.
@@ -582,7 +594,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', flags: 'MPLFlags', seq: 'int
 
 
 @schema_final
-class ILNPOption(Option):
+class ILNPOption(Option, code=Enum_Option.ILNP_Nonce):
     """Header schema for IPv6-Opts identifier-locator network protocol (ILNP) options."""
 
     #: Nonce value.
@@ -593,7 +605,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', nonce: 'int') -> 'None': ...
 
 
 @schema_final
-class LineIdentificationOption(Option):
+class LineIdentificationOption(Option, code=Enum_Option.Line_Identification_Option):
     """Header schema for IPv6-Opts line-identification options."""
 
     #: Line ID length.
@@ -606,7 +618,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', id_len: 'int', id: 'bytes')
 
 
 @schema_final
-class JumboPayloadOption(Option):
+class JumboPayloadOption(Option, code=Enum_Option.Jumbo_Payload):
     """Header schema for IPv6-Opts jumbo payload options."""
 
     #: Jumbo payload length.
@@ -617,7 +629,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', jumbo_len: 'int') -> 'None':
 
 
 @schema_final
-class HomeAddressOption(Option):
+class HomeAddressOption(Option, code=Enum_Option.Home_Address):
     """Header schema for IPv6-Opts home address options."""
 
     #: Home address.
@@ -628,7 +640,7 @@ def __init__(self, type: 'Enum_Option', len: 'int', addr: 'IPv6Address | int | s
 
 
 @schema_final
-class IPDFFOption(Option):
+class IPDFFOption(Option, code=Enum_Option.IP_DFF):
     """Header schema for IPv6-Opts depth-first forwarding (``IP_DFF``) options."""
 
     #: Flags.
@@ -657,23 +669,7 @@ class IPv6_Opts(Schema):
         length=lambda pkt: pkt['len'] * 8 + 6,
         base_schema=Option,
         type_name='type',
-        registry=collections.defaultdict(lambda: UnassignedOption, {
-            Enum_Option.Pad1: PadOption,
-            Enum_Option.PadN: PadOption,
-            Enum_Option.Tunnel_Encapsulation_Limit: TunnelEncapsulationLimitOption,
-            Enum_Option.Router_Alert: RouterAlertOption,
-            Enum_Option.CALIPSO: CALIPSOOption,
-            Enum_Option.SMF_DPD: _SMFDPDOption,
-            Enum_Option.PDM: PDMOption,
-            Enum_Option.Quick_Start: _QuickStartOption,
-            Enum_Option.RPL_Option_0x63: RPLOption,
-            Enum_Option.MPL_Option: MPLOption,
-            Enum_Option.ILNP_Nonce: ILNPOption,
-            Enum_Option.Line_Identification_Option: LineIdentificationOption,
-            Enum_Option.Jumbo_Payload: JumboPayloadOption,
-            Enum_Option.Home_Address: HomeAddressOption,
-            Enum_Option.IP_DFF: IPDFFOption,
-        })
+        registry=Option.registry,
     )
     #: Payload.
     payload: 'bytes' = PayloadField()