|
10 | 10 | from pcapkit.const.ipv6.router_alert import RouterAlert as Enum_RouterAlert |
11 | 11 | from pcapkit.const.ipv6.seed_id import SeedID as Enum_SeedID |
12 | 12 | from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode as Enum_SMFDPDMode |
13 | | -from pcapkit.const.ipv6.tagger_id import TaggerID as Enum_TaggerID |
14 | 13 | from pcapkit.const.reg.transtype import TransType as Enum_TransType |
15 | 14 | from pcapkit.corekit.fields.collections import OptionField |
16 | 15 | from pcapkit.corekit.fields.field import NO_VALUE |
17 | | -from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField |
18 | | -from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, NoValueField, |
19 | | - PayloadField, SchemaField, SwitchField) |
| 16 | +from pcapkit.corekit.fields.ipaddress import IPv6AddressField |
| 17 | +from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, PayloadField, |
| 18 | + SwitchField) |
20 | 19 | from pcapkit.corekit.fields.numbers import (EnumField, NumberField, UInt8Field, UInt16Field, |
21 | 20 | UInt32Field) |
22 | 21 | from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField |
| 22 | +from pcapkit.protocols.schema.internet import ipv6_option |
| 23 | +from pcapkit.protocols.schema.internet.ipv6_option import pad_opt_data_len, rpl_opt_sub_tlv_len |
23 | 24 | from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final |
24 | | -from pcapkit.utilities.exceptions import FieldValueError |
25 | 25 | from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING |
26 | 26 |
|
27 | 27 | __all__ = [ |
@@ -129,244 +129,83 @@ class QSNonce(TypedDict): |
129 | 129 | reserved: int |
130 | 130 |
|
131 | 131 |
|
132 | | -def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int': |
133 | | - """Return RPL option sub-TLV length. |
134 | | -
|
135 | | - Args: |
136 | | - pkt: RPL option unpacked schema. |
137 | | -
|
138 | | - Returns: |
139 | | - RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets |
140 | | - of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and |
141 | | - the reader then rejects the option. |
142 | | -
|
143 | | - """ |
144 | | - return max(pkt['len'] - 4, 0) |
| 132 | +# NOTE: The option helpers are shared with IPv6-Opts and live in |
| 133 | +# :mod:`pcapkit.protocols.schema.internet.ipv6_option`. The functions below bind |
| 134 | +# what differs between the two headers: the ``HOPOPT`` error prefix, and the |
| 135 | +# option schemas of this module. |
145 | 136 |
|
146 | 137 |
|
147 | 138 | def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int': |
148 | 139 | """Return MPL Seed-ID length. |
149 | 140 |
|
150 | | - Args: |
151 | | - pkt: MPL option unpacked schema. |
152 | | -
|
153 | | - Returns: |
154 | | - MPL Seed-ID length. |
| 141 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_seed_id_len`; |
| 142 | + errors are reported as ``HOPOPT``. |
155 | 143 |
|
156 | 144 | """ |
157 | | - s_type = pkt['flags']['type'] |
158 | | - if s_type == 0: |
159 | | - return 0 |
160 | | - if s_type == 1: |
161 | | - return 2 |
162 | | - if s_type == 2: |
163 | | - return 8 |
164 | | - if s_type == 3: |
165 | | - return 16 |
166 | | - raise FieldValueError(f'HOPOPT: invalid MPL Seed-ID type: {s_type}') |
| 145 | + return ipv6_option.mpl_opt_seed_id_len(pkt, prefix='HOPOPT') |
167 | 146 |
|
168 | 147 |
|
169 | 148 | def smf_i_dpd_id_len(pkt: 'dict[str, Any]') -> 'int': |
170 | 149 | """Return SMF I-DPD identifier length. |
171 | 150 |
|
172 | | - Args: |
173 | | - pkt: SMF identification-based DPD option unpacked schema. |
174 | | -
|
175 | | - Returns: |
176 | | - SMF I-DPD identifier length. |
177 | | -
|
178 | | - Raises: |
179 | | - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold |
180 | | - the TaggerID it declares, which would otherwise underflow the |
181 | | - identifier length below zero. |
| 151 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_id_len`; |
| 152 | + errors are reported as ``HOPOPT``. |
182 | 153 |
|
183 | 154 | """ |
184 | | - length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2)) |
185 | | - if length < 0: |
186 | | - raise FieldValueError(f'HOPOPT: invalid SMF I-DPD option length: {pkt["len"]}') |
187 | | - return length |
| 155 | + return ipv6_option.smf_i_dpd_id_len(pkt, prefix='HOPOPT') |
188 | 156 |
|
189 | 157 |
|
190 | 158 | def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field': |
191 | 159 | """Selector function for :attr:`_SMFDPDOption.data` field. |
192 | 160 |
|
193 | | - Args: |
194 | | - pkt: Packet data. |
195 | | -
|
196 | | - Returns: |
197 | | - * If ``mode`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` |
198 | | - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFIdentificationBasedDPDOption` |
199 | | - instance. |
200 | | - * If ``mode`` is ``1``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` |
201 | | - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.SMFHashBasedDPDOption` |
202 | | - instance. |
203 | | -
|
204 | | - Note: |
205 | | - The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``. |
206 | | - ``Opt Data Len`` counts only what follows the option header |
207 | | - [:rfc:`8200#section-4.2`], while both schemas this may return inherit |
208 | | - :attr:`Option.type` and :attr:`Option.len` and so parse those two octets |
209 | | - themselves. Sizing the field at ``Opt Data Len`` would hand them an area two |
210 | | - octets short of the option they read, which |
211 | | - :class:`~pcapkit.corekit.fields.collections.OptionField` would then |
212 | | - mis-count against the option area. |
| 161 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_dpd_data_selector`, |
| 162 | + choosing among the schemas registered with :class:`SMFDPDOption`; errors |
| 163 | + are reported as ``HOPOPT``. |
213 | 164 |
|
214 | 165 | """ |
215 | | - # NOTE: With ``Opt Data Len`` 0 the option has no data octet, so the mode |
216 | | - # bit forward-matched above belongs to whatever follows the option. |
217 | | - if pkt['test']['len'] == 0: |
218 | | - raise FieldValueError('HOPOPT: invalid SMF DPD option length: 0') |
219 | | - |
220 | | - mode = Enum_SMFDPDMode.get(pkt['test']['mode']) |
221 | | - schema = SMFDPDOption.registry[mode] |
222 | | - if schema is None: |
223 | | - raise FieldValueError(f'HOPOPT: invalid SMF DPD mode: {mode}') |
224 | | - return SchemaField(length=pkt['test']['len'] + 2, schema=schema) |
| 166 | + return ipv6_option.smf_dpd_data_selector(pkt, prefix='HOPOPT', base=SMFDPDOption) |
225 | 167 |
|
226 | 168 |
|
227 | 169 | def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field': |
228 | 170 | """Selector function for :attr:`SMFIdentificationBasedDPDOption.tid` field. |
229 | 171 |
|
230 | | - Args: |
231 | | - pkt: Packet data. |
232 | | -
|
233 | | - Returns: |
234 | | - * If ``tid_type`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.NoValueField` |
235 | | - instance. |
236 | | - * If ``tid_type`` is ``1``, returns a :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` |
237 | | - instance. |
238 | | - * If ``tid_type`` is ``2``, returns a :class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` |
239 | | - instance. |
240 | | - * Otherwise, returns a :class:`~pcapkit.corekit.fields.strings.BytesField` instance. |
| 172 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_tid_selector`; |
| 173 | + errors are reported as ``HOPOPT``. |
241 | 174 |
|
242 | 175 | """ |
243 | | - tid_type = Enum_TaggerID.get(pkt['info']['type']) |
244 | | - tid_len = pkt['info']['len'] |
245 | | - |
246 | | - # update type |
247 | | - pkt['info']['type'] = tid_type |
248 | | - |
249 | | - if tid_type == Enum_TaggerID.NULL: |
250 | | - if tid_len != 0: |
251 | | - raise FieldValueError(f'HOPOPT: invalid TaggerID length: {tid_len}') |
252 | | - return NoValueField() |
253 | | - if tid_type == Enum_TaggerID.IPv4: |
254 | | - if tid_len != 3: |
255 | | - raise FieldValueError(f'HOPOPT: invalid TaggerID length: {tid_len}') |
256 | | - return IPv4AddressField() |
257 | | - if tid_type == Enum_TaggerID.IPv6: |
258 | | - if tid_len != 15: |
259 | | - raise FieldValueError(f'HOPOPT: invalid TaggerID length: {tid_len}') |
260 | | - return IPv6AddressField() |
261 | | - return BytesField(length=tid_len + 1) |
| 176 | + return ipv6_option.smf_i_dpd_tid_selector(pkt, prefix='HOPOPT') |
262 | 177 |
|
263 | 178 |
|
264 | 179 | def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': |
265 | 180 | """Selector function for :attr:`_QuickStartOption.data` field. |
266 | 181 |
|
267 | | - Args: |
268 | | - pkt: Packet data. |
269 | | -
|
270 | | - Returns: |
271 | | - * If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` |
272 | | - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartRequestOption` |
273 | | - instance. |
274 | | - * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` |
275 | | - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartReportOption` |
276 | | - instance. |
277 | | - * Otherwise, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` |
278 | | - wrapped :class:`~pcapkit.protocols.schema.internet.hopopt.UnassignedOption` |
279 | | - instance, sized by the option's own ``Opt Data Len``. |
280 | | -
|
281 | | - """ |
282 | | - func = Enum_QSFunction.get(pkt['flags']['func']) |
283 | | - pkt['flags']['func'] = func |
284 | | - |
285 | | - schema = QuickStartOption.registry[func] |
286 | | - if schema is None: |
287 | | - # An unassigned function has no defined layout, so the option is kept |
288 | | - # as raw data, sized by its own ``Opt Data Len``. |
289 | | - return SchemaField(length=pkt['flags']['len'] + 2, schema=UnassignedOption) |
290 | | - |
291 | | - # NOTE: The suboption schemas carry the option's own ``type`` and ``len`` |
292 | | - # octets, so this is the width of the whole option: the two header octets |
293 | | - # plus the six data octets :rfc:`4782#section-3.2` gives both functions. |
294 | | - return SchemaField(length=8, schema=schema) |
295 | | - |
296 | | - |
297 | | -def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int': |
298 | | - """Return the length of the padding data of a padding option. |
299 | | -
|
300 | | - Args: |
301 | | - pkt: Padding option unpacked schema. |
302 | | -
|
303 | | - Returns: |
304 | | - Number of padding octets carried after the option header, i.e. the |
305 | | - value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for |
306 | | - a ``Pad1`` option, which carries no such field. |
307 | | -
|
308 | | - Note: |
309 | | - A ``Pad1`` option is a single octet with neither an ``Opt Data Len`` |
310 | | - field nor any option data (c.f. :rfc:`8200#section-4.2`), which is why |
311 | | - :attr:`Option.len` is declared as a |
312 | | - :class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped |
313 | | - for it. A skipped conditional field is *recorded* in the packet data as |
314 | | - :data:`~pcapkit.corekit.fields.field.NO_VALUE`, rather than being left |
315 | | - out of it, so the test below has to be on the **value** and not on the |
316 | | - presence of the key: ``pkt.get('len', 0)`` on its own hands that |
317 | | - :obj:`~pcapkit.corekit.fields.field.NoValueType` straight to |
318 | | - :class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes |
319 | | - an unusable :mod:`struct` template and surfaces much later as an opaque |
320 | | - :exc:`struct.error`, which is how a ``Pad1`` option would fail to parse. |
| 182 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector`, |
| 183 | + choosing among the schemas registered with :class:`QuickStartOption`, and |
| 184 | + :class:`UnassignedOption` for an unassigned QS function. |
321 | 185 |
|
322 | 186 | """ |
323 | | - length = pkt.get('len', 0) |
324 | | - if not isinstance(length, int): # ``NO_VALUE`` (skipped) or :obj:`None` (unset) |
325 | | - return 0 |
326 | | - return length |
| 187 | + return ipv6_option.quick_start_data_selector(pkt, base=QuickStartOption, |
| 188 | + unassigned=UnassignedOption) |
327 | 189 |
|
328 | 190 |
|
329 | 191 | def calipso_pad_len(pkt: 'dict[str, Any]') -> 'int': |
330 | 192 | """Return CALIPSO option padding length. |
331 | 193 |
|
332 | | - Args: |
333 | | - pkt: CALIPSO option unpacked schema. |
334 | | -
|
335 | | - Returns: |
336 | | - CALIPSO option padding length. |
337 | | -
|
338 | | - Raises: |
339 | | - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold |
340 | | - the fixed header and the compartment bitmap declared by |
341 | | - ``cmpt_len``, which would otherwise underflow the padding length |
342 | | - below zero. |
| 194 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len`; |
| 195 | + errors are reported as ``HOPOPT``. |
343 | 196 |
|
344 | 197 | """ |
345 | | - length = pkt['len'] - 8 - pkt['cmpt_len'] * 4 |
346 | | - if length < 0: |
347 | | - raise FieldValueError(f'HOPOPT: invalid CALIPSO option length: {pkt["len"]}') |
348 | | - return length |
| 198 | + return ipv6_option.calipso_pad_len(pkt, prefix='HOPOPT') |
349 | 199 |
|
350 | 200 |
|
351 | 201 | def mpl_opt_pad_len(pkt: 'dict[str, Any]') -> 'int': |
352 | 202 | """Return MPL option padding length. |
353 | 203 |
|
354 | | - Args: |
355 | | - pkt: MPL option unpacked schema. |
356 | | -
|
357 | | - Returns: |
358 | | - MPL option padding length. |
359 | | -
|
360 | | - Raises: |
361 | | - FieldValueError: If ``Opt Data Len`` on the wire is too short to hold |
362 | | - the fixed header and the Seed-ID declared by ``flags.type``, which |
363 | | - would otherwise underflow the padding length below zero. |
| 204 | + See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_pad_len`; |
| 205 | + errors are reported as ``HOPOPT``. |
364 | 206 |
|
365 | 207 | """ |
366 | | - length = pkt['len'] - 2 - (0 if pkt['flags']['type'] == 0 else mpl_opt_seed_id_len(pkt)) |
367 | | - if length < 0: |
368 | | - raise FieldValueError(f'HOPOPT: invalid MPL option length: {pkt["len"]}') |
369 | | - return length |
| 208 | + return ipv6_option.mpl_opt_pad_len(pkt, prefix='HOPOPT') |
370 | 209 |
|
371 | 210 |
|
372 | 211 | class Option(EnumSchema[Enum_Option]): |
|
0 commit comments