PyPCAPKit

https://jarryshaw.github.io/PyPCAPKit
Admin
PyPCAPKit is an open-source Python library for comprehensive network packet parsing, PCAP file...
Tokens:659,087
Snippets:3,188
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install PyPCAPKit from Source

Source: https://jarryshaw.github.io/PyPCAPKit/index.html

Commands to clone the repository and perform an editable installation for development.

```bash
git clone https://github.com/JarryShaw/PyPCAPKit.git
cd pypcapkit
pip install -e .
# and to update at any time
git pull
```

--------------------------------

### Define Quick Start Options

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/data/internet/ipv6_opts.html

Data models for IPv6-Opts Quick Start request and report options.

```python
class QuickStartOption(Option):
    """Data model for IPv6-Opts Quick Start option."""

    #: QS function.
    func: 'QSFunction'
    #: Rate request/report.
    rate: 'int'



@info_final
class QuickStartRequestOption(QuickStartOption):
    """Data model for IPv6-Opts Quick Start request option."""

    #: TTL.
    ttl: 'timedelta'
    #: Nonce.
    nonce: 'int'

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', func: 'QSFunction', rate: 'int', ttl: 'timedelta',
                     nonce: 'int') -> 'None': ...  # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long


@info_final
class QuickStartReportOption(QuickStartOption):
    """Data model for IPv6-Opts Quick Start report of approved rate option."""

    #: Nonce.
    nonce: 'int'

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', func: 'QSFunction', rate: 'int', nonce: 'int') -> 'None':
            ...
```

--------------------------------

### Install PyPCAPKit via PyPI

Source: https://jarryshaw.github.io/PyPCAPKit/index.html

Standard installation command for the current version of the library.

```bash
pip install pypcapkit
```

--------------------------------

### IPv4 Quick Start Parsing Logic

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/ipv4.html

Implementation for parsing Quick Start options, handling both Request and Report functions.

```python
if schema.length != 8:
    raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format')

func = schema.func
if func == Enum_QSFunction.Quick_Start_Request:
    schema_req = cast('Schema_QuickStartRequestOption', schema)

    rate = schema_req.flags['rate']
    opt = Data_QuickStartRequestOption(
        code=schema.type,
        type=self._read_ipv4_opt_type(schema.type),
        length=schema_req.length,
        func=func,
        rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0,
        ttl=datetime.timedelta(seconds=schema_req.ttl),
        nonce=schema_req.nonce['nonce'],
    )  # type: Data_QSOption
elif func == Enum_QSFunction.Report_of_Approved_Rate:
    schema_rep = cast('Schema_QuickStartReportOption', schema)

    rate = schema_rep.flags['rate']
    opt = Data_QuickStartReportOption(
        code=schema.type,
        type=self._read_ipv4_opt_type(schema.type),
        length=schema_rep.length,
        func=func,
        rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0,
        nonce=schema_rep.nonce['nonce'],
    )
else:
    raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] unknown QS function: {func}')
return opt
```

--------------------------------

### Define Quick Start option data models

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/data/internet/ipv4.html

Data models for IPv4 Quick Start (QS) options, including request and report variants.

```python
[docs]
class QSOption(Option):
    """Data model for IPv4 Quick Start (``QS``) option."""

    #: QS function.
    func: 'QSFunction'
    #: Rate request/report.
    rate: 'int'




@info_final
class QuickStartRequestOption(QSOption):
    """Data model for IPv4 Quick Start request option."""

    #: TTL.
    ttl: 'timedelta'
    #: Nonce.
    nonce: 'int'

    if TYPE_CHECKING:
         def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', func: 'QSFunction', rate: 'int', ttl: 'timedelta', nonce: 'int') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin


@info_final
class QuickStartReportOption(QSOption):
    """Data model for IPv4 Quick Start report of approved rate option."""

    #: Nonce.
    nonce: 'int'

    if TYPE_CHECKING:
        def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', func: 'QSFunction', rate: 'int', nonce: 'int') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
```

--------------------------------

### Define IPv4 Quick Start Option Schemas

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/schema/internet/ipv4.html

Defines the base and specific schemas for IPv4 quick start (QS) options.

```python
@schema_final
class _QSOption(Schema):
    """Header schema for IPv4 quick start (``QS``) options in generic representation."""

    #: Flags.
    flags: 'QSTestFlags' = ForwardMatchField(BitField(length=3, namespace={
        'func': (16, 4),
    }))
    #: QS data.
    data: 'QuickStartRequestOption | QuickStartReportOption' = SwitchField(
        selector=quick_start_data_selector,
    )

    def post_process(self, packet: 'dict[str, Any]') -> 'QSOption':
        """Revise ``schema`` data after unpacking process.

        Args:
            packet: Unpacked data.

        Returns:
            Revised schema.

        """
        ret = self.data
        ret.func = Enum_QSFunction.get(self.flags['func'])
        return ret


# register ``_QSOption`` as ``QS`` option
Option.register(Enum_OptionNumber.QS, _QSOption)


class QSOption(Option, EnumSchema[Enum_QSFunction]):
    """Header schema for IPV4 quick start (``QS``) options."""

    __enum__: 'DefaultDict[Enum_QSFunction, Type[QSOption]]' = collections.defaultdict(lambda: None)  # type: ignore[return-value,arg-type]

    #: Flags.
    flags: 'QuickStartFlags' = BitField(length=1, namespace={
        'func': (0, 4),
        'rate': (4, 4),
    })

    if TYPE_CHECKING:
        func: 'Enum_QSFunction'


@schema_final
class QuickStartRequestOption(QSOption, code=Enum_QSFunction.Quick_Start_Request):
    """Header schema for IPV4 quick start request options."""

    #: QS time-to-live (TTL).
    ttl: 'int' = UInt8Field()
    #: QS nonce.
    nonce: 'QSNonce' = BitField(length=4, namespace={
        'nonce': (0, 30),
    })

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_OptionNumber', length: 'int', flags: 'QuickStartFlags',
                     ttl: 'int', nonce: 'QSNonce') -> 'None': ...


@schema_final
class QuickStartReportOption(QSOption, code=Enum_QSFunction.Report_of_Approved_Rate):
    """Header schema for IPV4 quick start report of approved rate options."""

    #: QS nonce.
    nonce: 'QSNonce' = BitField(length=4, namespace={
        'nonce': (0, 30),
    })

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_OptionNumber', length: 'int', flags: 'QuickStartFlags',
                     nonce: 'QSNonce') -> 'None': ...
```

--------------------------------

### Install CLI Dependencies

Source: https://jarryshaw.github.io/PyPCAPKit/index.html

Commands to install the necessary dependencies for command-line interface usage.

```bash
pip install pypcapkit[cli]
# or explicitly...
pip install pypcapkit emoji
```

--------------------------------

### Define QuickStartOption schema

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/schema/internet/hopopt.html

Defines the base schema for HOPOPT quick start options and registers it.

```python
@schema_final
class _QuickStartOption(Schema):
    """Header schema for HOPOPT quick start options in generic representation."""

    #: Flags.
    flags: 'QSTestFlags' = ForwardMatchField(BitField(length=3, namespace={
        'func': (16, 4),
    }))
    #: QS data.
    data: 'QuickStartRequestOption | QuickStartReportOption' = SwitchField(
        selector=quick_start_data_selector,
    )



    def post_process(self, packet: 'dict[str, Any]') -> 'QuickStartOption':
        """Revise ``schema`` data after unpacking process.

        Args:
            packet: Unpacked data.

        Returns:
            Revised schema.

        """
        ret = self.data
        ret.func = Enum_QSFunction.get(self.flags['func'])
        return ret




# register ``_QuickStartOption`` as ``Quick_Start`` option
Option.register(Enum_Option.Quick_Start, _QuickStartOption)
```

--------------------------------

### Quick Start Option Structure

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/hopopt.html

Visual representation of the Quick Start Request and Report of Approved Rate options.

```text
0                   1                   2                   3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|   Option      |  Length=6     | Func. | Rate  |   QS TTL      |
|               |               | 0000  |Request|               |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                        QS Nonce                           | R |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
```

```text
0                   1                   2                   3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|   Option      |  Length=6     | Func. | Rate  |   Not Used    |
|               |               | 1000  | Report|               |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                        QS Nonce                           | R |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
```

--------------------------------

### Define QuickStartOption Schema

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/schema/internet/ipv6_opts.html

Defines the base and specific Quick Start option schemas, including request and report variants with registration.

```python
@schema_final
class _QuickStartOption(Schema):
    """Header schema for IPv6-Opts quick start options in generic representation."""

    #: Flags.
    flags: 'QSTestFlags' = ForwardMatchField(BitField(length=3, namespace={
        'func': (16, 4),
    }))
    #: QS data.
    data: 'QuickStartRequestOption | QuickStartReportOption' = SwitchField(
        selector=quick_start_data_selector,
    )

    def post_process(self, packet: 'dict[str, Any]') -> 'QuickStartOption':
        """Revise ``schema`` data after unpacking process.

        Args:
            packet: Unpacked data.

        Returns:
            Revised schema.

        """
        ret = self.data
        ret.func = Enum_QSFunction.get(self.flags['func'])
        return ret


# 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.
    flags: 'QuickStartFlags' = BitField(length=1, namespace={
        'func': (0, 4),
        'rate': (4, 4),
    })

    if TYPE_CHECKING:
        func: 'Enum_QSFunction'


@schema_final
class QuickStartRequestOption(QuickStartOption, code=Enum_QSFunction.Quick_Start_Request):
    """Header schema for IPv6-Opts quick start request options."""

    #: QS time-to-live (TTL).
    ttl: 'int' = UInt8Field()
    #: QS nonce.
    nonce: 'QSNonce' = BitField(length=4, namespace={
        'nonce': (0, 30),
    })

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
                     ttl: 'int', nonce: 'QSNonce') -> 'None': ...


@schema_final
class QuickStartReportOption(QuickStartOption, code=Enum_QSFunction.Report_of_Approved_Rate):
    """Header schema for IPv6-Opts quick start report of approved rate options."""

    #: Reserved.
    reserved: 'bytes' = PaddingField(length=1)
    #: QS nonce.
    nonce: 'QSNonce' = BitField(length=4, namespace={
        'nonce': (0, 30),
    })

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
                     nonce: 'QSNonce') -> 'None': ...
```

--------------------------------

### Define QuickStartRequest and Report options

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/schema/internet/hopopt.html

Defines specific schema classes for Quick Start request and report options.

```python
@schema_final
class QuickStartRequestOption(QuickStartOption, code=Enum_QSFunction.Quick_Start_Request):
    """Header schema for HOPOPT quick start request options."""

    #: QS time-to-live (TTL).
    ttl: 'int' = UInt8Field()
    #: QS nonce.
    nonce: 'QSNonce' = BitField(length=4, namespace={
        'nonce': (0, 30),
    })

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
                     ttl: 'int', nonce: 'QSNonce') -> 'None': ...
```

```python
@schema_final
class QuickStartReportOption(QuickStartOption, code=Enum_QSFunction.Report_of_Approved_Rate):
    """Header schema for HOPOPT quick start report of approved rate options."""

    #: Reserved.
    reserved: 'bytes' = PaddingField(length=1)
    #: QS nonce.
    nonce: 'QSNonce' = BitField(length=4, namespace={
        'nonce': (0, 30),
    })

    if TYPE_CHECKING:
        def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
                     nonce: 'QSNonce') -> 'None': ...
```

--------------------------------

### IPv4 Quick Start Option Structures

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/ipv4.html

Visual representations of the Quick-Start Request and Report of Approved Rate options as defined in RFC 4782.

```text
0                   1                   2                   3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|   Option      |  Length=8     | Func. | Rate  |   QS TTL      |
|               |               | 0000  |Request|               |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                        QS Nonce                           | R |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
```

```text
0                   1                   2                   3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|   Option      |  Length=8     | Func. | Rate  |   Not Used    |
|               |               | 1000  | Report|               |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                        QS Nonce                           | R |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
```

--------------------------------

### Extractor Initialization and Setup

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/foundation/extraction.html

Internal logic for configuring trace flow managers, input file streams, and output dumpers during Extractor initialization.

```python
trace_cls_tcp = self.__traceflow__['tcp']
if isinstance(trace_cls_tcp, ModuleDescriptor):
    trace_cls_tcp = trace_cls_tcp.klass
    self.__traceflow__['tcp'] = trace_cls_tcp  # update mapping upon import
trace_obj_tcp = cast('TCP_TraceFlow', trace_cls_tcp(fout=trace_fout, format=trace_format,
                                                    byteorder=trace_byteorder, nanosecond=trace_nanosecond))

self._trace = TraceFlowManager(
    tcp=trace_obj_tcp,
)

if self._flag_s:
    self._ifile = open(ifnm, 'rb')  # input file # pylint: disable=unspecified-encoding,consider-using-with
else:
    self._ifile = cast('BufferedReader', fin)

if not self._ifile.seekable():
    self._ifile = SeekableReader(self._ifile, buffer_size, buffer_save, buffer_path,
                                 stream_closing=not self._flag_s)

if not self._flag_q:
    output, ext = self.__output__[fmt]
    if ext is None:
        warn(f'Unsupported output format: {fmt}; disabled file output feature',
             FormatWarning, stacklevel=stacklevel())
    if isinstance(output, ModuleDescriptor):
        output = output.klass
        self.__output__[fmt] = (output, ext)  # update mapping upon import
    dumper = make_dumper(output)

    self._ofile = dumper if self._flag_f else dumper(ofnm)  # output file

# NOTE: we use peek() to read the magic number, as the file pointer
# will not be moved after reading; however, the returned bytes object
# may not be exactly 4 bytes, so we use [:4] to get the first 4 bytes
self._magic = self._ifile.peek(4)[:4]
#self._magic = self._ifile.read(4)  # magic number
#self._ifile.seek(0, os.SEEK_SET)

self.run()    # start extraction
```

--------------------------------

### CLI Export Samples

Source: https://jarryshaw.github.io/PyPCAPKit/demo.html

Examples of using the pcapkit-cli command to export PCAP files to different formats.

```bash
$ pcapkit-cli in --format plist --verbose
🚨Loading file 'in.pcap'
Frame   1: Ethernet:IPv6:IPv6_ICMP
Frame   2: Ethernet:IPv6:IPv6_ICMP
Frame   3: Ethernet:IPv4:TCP
Frame   4: Ethernet:IPv4:TCP
Frame   5: Ethernet:IPv4:TCP
Frame   6: Ethernet:IPv4:UDP:Raw
🍺Report file stored in 'out.plist'
```

```bash
$ pcapkit-cli in --output out.json --verbose
🚨Loading file 'in.pcap'
Frame   1: Ethernet:IPv6:IPv6_ICMP
Frame   2: Ethernet:IPv6:IPv6_ICMP
Frame   3: Ethernet:IPv4:TCP
Frame   4: Ethernet:IPv4:TCP
Frame   5: Ethernet:IPv4:TCP
Frame   6: Ethernet:IPv4:UDP:Raw
🍺Report file stored in 'out.json'
```

```bash
$ pcapkit-cli in --output out.txt --format tree --verbose
🚨Loading file 'in.pcap'
Frame   1: Ethernet:IPv6:IPv6_ICMP
Frame   2: Ethernet:IPv6:IPv6_ICMP
Frame   3: Ethernet:IPv4:TCP
Frame   4: Ethernet:IPv4:TCP
Frame   5: Ethernet:IPv4:TCP
Frame   6: Ethernet:IPv4:UDP:Raw
🍺Report file stored in 'out.txt'
```

--------------------------------

### Construct TCP Quick Start Response option schema

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/transport/tcp.html

Generates a Schema_QuickStartResponse object, performing logarithmic rate conversion and time difference calculation.

```python
    def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartResponse]' = None, *,
                      rate: 'int' = 0,
                      diff: 'timedelta | int' = 0,
                      nonce: 'int' = 0,
                      **kwargs: 'Any') -> 'Schema_QuickStartResponse':
        """Make TCP quick start response option.

        Args:
            code: option code
            opt: option data
            rate: rate (in kbps)
            diff: time to live (in seconds) difference
            nonce: nonce value
            **kwargs: arbitrary keyword arguments

        Returns:
            Constructed option schema.

        """
        if opt is not None:
            rate = opt.req_rate
            diff = opt.ttl_diff
            nonce = opt.nonce

        rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0
        diff_val = diff if isinstance(diff, int) else math.floor(diff.total_seconds())

        return Schema_QuickStartResponse(
            kind=code,
            length=8,
            flags={
                'rate': rate_val,
            },
            diff=diff_val,
            nonce={
                'nonce': nonce,
            },
        )
```

--------------------------------

### Initialize HTTP Data

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/application/httpv1.html

Example of initializing a Data_HTTP object with receipt, header, and body components.

```python
        http = Data_HTTP(
            receipt=header_line,
            header=header_unpacked,
            body=body_unpacked,
        )
        self._receipt = header_line.type
        self._version = header_line.version  # type: ignore[attr-defined]
        self._length = len(header)

        return http
```

--------------------------------

### Construct Quick-Start Option

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/ipv4.html

Creates an IPv4 Quick-Start (QS) option schema, supporting both request and report functions based on the provided option data or parameters.

```python
    def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickStartRequestOption | Data_QuickStartReportOption]' = None, *,
                     func: 'Enum_QSFunction | StdlibEnum | AenumEnum | str | int' = Enum_QSFunction.Quick_Start_Request,
                     func_default: 'Optional[int]' = None,
                     func_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None,   # pylint: disable=line-too-long
                     func_reversed: 'bool' = False,
                     rate: 'int' = 0,
                     ttl: 'timedelta | int' = 0,
                     nonce: 'int' = 0,
                     **kwargs: 'Any') -> 'Schema_QSOption':
        """Make IPv4 Quick-Start (``QS``) option.

        Args:
            code: option type value
            opt: option data
            func: QS function type
            func_default: default value for QS function type
            func_namespace: namespace for QS function type
            func_reversed: reversed flag for QS function type
            rate: rate (in kbps)
            ttl: time to live (in seconds)
            nonce: nonce value
            **kwargs: arbitrary keyword arguments

        Returns:
            Constructured option schema.

        """
        if option is not None:
            func_enum = option.func
            rate = option.rate
            ttl = getattr(option, 'ttl', 0)
            nonce = option.nonce
        else:
            func_enum = self._make_index(func, func_default, namespace=func_namespace,  # type: ignore[assignment]
                                         reversed=func_reversed, pack=False)
        rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0

        if func_enum == Enum_QSFunction.Quick_Start_Request:
            ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds())

            return Schema_QuickStartRequestOption(
                type=kind,
                length=8,
                flags={
                    'func': func_enum,
                    'rate': rate_val,
                },
                ttl=ttl_value,
                nonce={
                    'nonce': nonce,
                },
            )
        if func_enum == Enum_QSFunction.Report_of_Approved_Rate:
            return Schema_QuickStartReportOption(
                type=kind,
                length=8,
                flags={
                    'func': func_enum,
                    'rate': rate_val,
                },
                nonce={
                    'nonce': nonce,
                },
            )
        raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid QS function: {func_enum}')
```

--------------------------------

### Install Optional Engine Dependencies

Source: https://jarryshaw.github.io/PyPCAPKit/index.html

Commands to install specific extraction engines or the full suite of optional packages.

```bash
# for DPKT only
pip install pypcapkit[DPKT]
# for Scapy only
pip install pypcapkit[Scapy]
# for PyShark only
pip install pypcapkit[PyShark]
# and to install all the optional packages
pip install pypcapkit[all]
# or to do this explicitly
pip install pypcapkit dpkt scapy pyshark
```

--------------------------------

### Read Quick Start Response Data

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/transport/tcp.html

Processes the Quick Start Response schema to calculate request rate and return structured data.

```python
        rate = schema.flags['rate']
        data = Data_QuickStartResponse(
            kind=schema.kind,
            length=schema.length,
            req_rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0,
            ttl_diff=schema.diff,
            nonce=schema.nonce['nonce'],
        )
        return data
```

--------------------------------

### Parse Quick-Start Response Option

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/transport/tcp.html

Validates the QSopt option length and performs initial byte unpacking.

```python
size = self._read_unpack(1)
if schema.length != 8:
    raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
```

--------------------------------

### Get PCAP-NG payload

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/misc/pcapng.html

Retrieves the payload from the block header.

```python
    def _get_payload(self) -> 'bytes':
        """Get payload of :attr:`self.__header__ <pcapkit.protocols.protocol.Protocol.__header__>`.

        Returns:
            Payload of :attr:`self.__header__ <pcapkit.protocols.protocol.Protocol.__header__>` as :obj:`bytes`.

        See Also:
            This is a wrapper function for :meth:`pcapkit.protocols.schema.Schema.get_payload`.

        """
        return self.__header__.block.get_payload()
```

--------------------------------

### run

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/foundation/extraction.html

Starts the packet extraction process using the configured engine.

```APIDOC
## run()

### Description
Starts the extraction process. It checks for engine availability and falls back to default engines if the specified engine is not installed or supported.
```

--------------------------------

### Implement Quick-Start data selector

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/schema/internet/ipv4.html

Selector function to determine the appropriate schema for Quick-Start options based on the function flag.

```python
def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field':
    """Selector function for :attr:`_QSOption.data` field.

    Args:
        pkt: Packet data.

    Returns:
        * If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
          wrapped :class:`~pcapkit.protocols.schema.internet.ipv4.QuickStartRequestOption`
          instance.
        * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
          wrapped :class:`~pcapkit.protocols.schema.internet.ipv4.QuickStartReportOption`
          instance.

    """
    func = Enum_QSFunction.get(pkt['flags']['func'])
    pkt['flags']['func'] = func

    schema = QSOption.registry[func]
    if schema is None:
        raise FieldValueError(f'IPv4: invalid QS function: {func}')
    return SchemaField(length=5, schema=schema)
```

--------------------------------

### Initialize HOPOPT protocol instance

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/hopopt.html

Post-initialization hook for setting up the HOPOPT protocol instance, specifically handling extension header flags.

```python
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *,  # pylint: disable=arguments-differ
                      extension: 'bool' = False, **kwargs: 'Any') -> 'None':
        """Post initialisation hook.

        Args:
            file: Source packet stream.
            length: Length of packet data.
            extension: If the protocol is used as an IPv6 extension header.
            **kwargs: Arbitrary keyword arguments.

        See Also:
            For construction argument, please refer to :meth:`make`.

        """
        #: bool: If the protocol is used as an IPv6 extension header.
        self._extf = extension

        # call super __post_init__
        super().__post_init__(file, length, extension=extension, **kwargs)  # type: ignore[arg-type]
```

--------------------------------

### EtherType Registry Constants

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/const/reg/ethertype.html

Example definitions for specific EtherType constants.

```python
BBN_VITAL_LanBridge_cache_private_protocol = 0xFF00

#: Reserved [:rfc:`1701`]
Reserved = 0xFFFF
```

--------------------------------

### Get File Descriptor

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/corekit/io.html

Returns the underlying integer file descriptor of the stream.

```python
def fileno(self) -> 'int':
    """Return the underlying file descriptor (an integer) of the stream if it exists.
    An :exc:`OSError` is raised if the IO object does not use a file descriptor."""
    return self._stream.fileno()
```

--------------------------------

### Construct HOPOPT QS Option

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/hopopt.html

Defines the interface for creating a Quick-Start option schema with various configuration parameters.

```python
    def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption]' = None, *,
                     func: 'Enum_QSFunction | StdlibEnum | AenumEnum | str | int' = Enum_QSFunction.Quick_Start_Request,
                     func_default: 'Optional[int]' = None,
                     func_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None,   # pylint: disable=line-too-long
                     func_reversed: 'bool' = False,
                     rate: 'int' = 0,
                     ttl: 'timedelta | int' = 0,
                     nonce: 'int' = 0,
                     **kwargs: 'Any') -> 'Schema_QuickStartOption':
        """Make HOPOPT QS option.

        Args:
            code: option type value
            opt: option data
            func: QS function type
            func_default: default value for QS function type
            func_namespace: namespace for QS function type
            func_reversed: reversed flag for QS function type
            rate: rate (in kbps)
            ttl: time to live (in seconds)
            nonce: nonce value
            **kwargs: arbitrary keyword arguments

        Returns:
            Constructured option schema.

```

--------------------------------

### Construct MH Schema

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/mh.html

Example of returning a Schema_MH object with calculated length.

```python
        return Schema_MH(
            next=next_val,
            length=math.ceil((len(data_val) + 6) / 8) - 1,
            type=type_val,
            chksum=chksum,
            data=data_val,
            payload=payload,
        )
```

--------------------------------

### Execute HandoverInitiateFlag Script

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/vendor/mh/handover_initiate_flag.html

Entry point for executing the module directly.

```python
if __name__ == '__main__':
    sys.exit(HandoverInitiateFlag())  # type: ignore[arg-type]
```

--------------------------------

### Representing Protocol Data

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/protocol.html

Example of the string representation returned by the protocol object.

```python
>>> protocol
<Frame alias='...' frame=(..., packet=b'...', sethernet=..., protocols='Ethernet:IPv6:Raw')>
```

--------------------------------

### run()

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/foundation/engines/scapy.html

Starts the packet extraction process using Scapy's sniff functionality.

```APIDOC
## Method: run()

### Description
Calls scapy.sendrecv.sniff to extract packets from the PCAP file specified in the extractor. Note that layer and protocol thresholds are not supported by the Scapy engine and will be ignored if provided.
```

--------------------------------

### Get ARP Protocol Index

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/link/arp.html

Retrieves the IANA registry index for the ARP protocol.

```python
@classmethod
def __index__(cls) -> 'Enum_EtherType':  # pylint: disable=invalid-index-returned
    """Numeral registry index of the protocol.

    Returns:
        Numeral registry index of the protocol in `IANA`_.

    .. _IANA: https://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml

    """
    return Enum_EtherType.Address_Resolution_Protocol  # type: ignore[return-value]
```

--------------------------------

### IPv6.__init__

Source: https://jarryshaw.github.io/PyPCAPKit/pcapkit/foundation/reassembly/ip/ipv6.html

Initializes the IPv6 reassembly instance.

```APIDOC
## IPv6(strict=True, store=True)

### Description
Initializes a new IPv6 reassembly object.

### Parameters
- **strict** (bool) - Optional - If True, returns all datagrams (including those not implemented) when submitted.
- **store** (bool) - Optional - If True, stores the reassembled datagram in memory.
```

--------------------------------

### Get ARP Packet Length Hint

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/link/arp.html

Returns the estimated length of the ARP object.

```python
def __length_hint__(self) -> 'Literal[28]':
    """Return an estimated length for the object."""
    return 28
```

--------------------------------

### Get datagram index

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/foundation/reassembly/reassembly.html

Finds the index of a datagram containing a specific packet number.

```python
def index(self, pkt_num: 'int') -> 'Optional[int]':
    """Return datagram index.

    Arguments:
        pkt_num: index of packet

    Returns:
        Reassembled datagram index which was from No. ``pkt_num`` packet;
        if not found, returns :obj:`None`.

    """
    for counter, datagram in enumerate(self.datagram):
        if pkt_num in datagram.index:  # type: ignore[attr-defined]
            return counter
    return None
```

--------------------------------

### Initialize IPv6 Options Protocol

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/ipv6_opts.html

The __post_init__ method handles protocol initialization, setting the extension flag and calling the superclass initializer.

```python
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *,  # pylint: disable=arguments-differ
                      extension: 'bool' = False, **kwargs: 'Any') -> 'None':
        """Post initialisation hook.

        Args:
            file: Source packet stream.
            length: Length of packet data.
            extension: If the protocol is used as an IPv6 extension header.
            **kwargs: Arbitrary keyword arguments.

        See Also:
            For construction argument, please refer to :meth:`self.make <IPv6_Opts.make>`.

        """
        #: bool: If the protocol is used as an IPv6 extension header.
        self._extf = extension

        # call super __post_init__
        super().__post_init__(file, length, extension=extension, **kwargs)  # type: ignore[arg-type]
```

--------------------------------

### Construct IPv6-Opts QS Option

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/ipv6_opts.html

Defines the interface for creating a Quick-Start option schema with parameters for rate, TTL, and nonce.

```python
    def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption]' = None, *,
                     func: 'Enum_QSFunction | StdlibEnum | AenumEnum | str | int' = Enum_QSFunction.Quick_Start_Request,
                     func_default: 'Optional[int]' = None,
                     func_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None,   # pylint: disable=line-too-long
                     func_reversed: 'bool' = False,
                     rate: 'int' = 0,
                     ttl: 'timedelta | int' = 0,
                     nonce: 'int' = 0,
                     **kwargs: 'Any') -> 'Schema_QuickStartOption':
        """Make IPv6-Opts QS option.

        Args:
            code: option type value
            opt: option data
            func: QS function type
            func_default: default value for QS function type
            func_namespace: namespace for QS function type
            func_reversed: reversed flag for QS function type
            rate: rate (in kbps)
            ttl: time to live (in seconds)
            nonce: nonce value
            **kwargs: arbitrary keyword arguments

        Returns:
            Constructured option schema.

```

--------------------------------

### Initialize schema fields

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/schema/schema.html

The __post_init__ method populates default values and triggers the initial packing process.

```python
def __post_init__(self, packet: 'Optional[dict[str, Any]]' = None) -> 'None':
    for name, field in self.__fields__.items():
        if self.__dict__[name] in (NoValue, None):
            self.__dict__[name] = field.default
    self.pack(packet)
```

--------------------------------

### Seek Position

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/corekit/io.html

Changes the stream position relative to the start, current position, or end of the stream.

```python
def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET, /) -> 'int':
    """Change the stream position to the given byte ``offset``. ``offset`` is interpreted
    relative to the position indicated by ``whence``. The default value for ``whence`` is
    :data:`~io.SEEK_SET`. Values for ``whence`` are:

    * :data:`~io.SEEK_SET` or ``0`` - start of the stream (the default); ``offset`` should
      be zero or positive
    * :data:`~io.SEEK_CUR` or ``1`` - current stream position; ``offset`` may be negative
    * :data:`~io.SEEK_END` or ``2`` - end of the stream; ``offset`` is usually negative

    Return the new absolute position.

    """
    # NOTE: we mark the end of buffer content to the end of buffer
    # so that it may trigger the IO to read more data to fill in
    # the content.
    buf_end = self._buffer_set + self._buffer_size
    #buf_end = self._buffer_set + self._buffer_cur
```

--------------------------------

### Protocol object initialization

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/protocol.html

Handles the creation and initialization of protocol instances, including setting up internal state and post-initialization hooks.

```python
    def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'Self':  # pylint: disable=unused-argument
        self = super().__new__(cls)

        # NOTE: Assign this attribute after ``__new__`` to avoid shared memory
        # reference between instances.
        self.__cached__ = {}
        self.__header__ = None  # type: ignore[assignment]

        return self

    @overload
    def __init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., **kwargs: 'Any') -> 'None': ...
    @overload
    def __init__(self, **kwargs: 'Any') -> 'None': ...

    def __init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'None':
        """Initialisation.

        Args:
            file: Source packet stream.
            length: Length of packet data.
            _layer (str): Parse packet until ``_layer``
                (:attr:`self._exlayer <pcapkit.protocols.protocol.Protocol._exlayer>`).
            _protocol (Union[str, Protocol, Type[Protocol]]): Parse packet until ``_protocol``
                (:attr:`self._exproto <pcapkit.protocols.protocol.Protocol._exproto>`).
            **kwargs: Arbitrary keyword arguments.

        """
        #logger.debug('%s(file, %s, **%s)', type(self).__name__, length, kwargs)

        #: int: File pointer.
        self._seekset = io.SEEK_SET  # type: int
        #: str: Parse packet until such layer.
        self._exlayer = kwargs.pop('_layer', None)  # type: Optional[str]
        #: str: Parse packet until such protocol.
        self._exproto = kwargs.pop('_protocol', None)  # type: Optional[str | ProtocolBase | Type[ProtocolBase]]
        #: bool: If terminate parsing next layer of protocol.
        self._sigterm = self._check_term_threshold()

        # post-init customisations
        self.__post_init__(file, length, **kwargs)  # type: ignore[arg-type]
```

--------------------------------

### Formatting Hex Representation

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/protocol.html

Example of the formatted hex and string output generated by the protocol object.

```python
>>> protocol
Frame(..., packet=b"...", sethernet=..., protocols='Ethernet:IPv6:Raw')
>>> print(protocol)
00 00 00 00 00 00 00 a6 87 f9 27 93 16 ee fe 80 00 00 00     ..........'........
00 00 00 1c cd 7c 77 ba c7 46 b7 87 00 0e aa 00 00 00 00     .....|w..F.........
fe 80 00 00 00 00 00 00 1c cd 7c 77 ba c7 46 b7 01 01 a4     ..........|w..F....
5e 60 d9 6b 97                                               ^`.k.
```

--------------------------------

### PCAPIO.__init__

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/dumpkit/pcap.html

Initializes a new PCAPIO dumper instance for writing PCAP files.

```APIDOC
## PCAPIO.__init__

### Description
Initializes the dumper with the specified output file, data link type, and configuration settings.

### Parameters
- **fname** (str) - Required - Output file name.
- **protocol** (Enum_LinkType | StdlibIntEnum | AenumIntEnum | str | int) - Required - Data link type.
- **byteorder** (Literal["big", "little"]) - Optional - Header byte order (defaults to system byte order).
- **nanosecond** (bool) - Optional - Nanosecond-resolution file flag (defaults to False).
- ****kwargs** (Any) - Optional - Arbitrary keyword arguments.
```

--------------------------------

### Define FTP Command Constants

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/const/ftp/command.html

Examples of FTP command constant definitions mapping to RFC specifications.

```python
STAT: 'Command' = 'STAT', FEATCode.base, 'Status', CommandType.S, ConformanceRequirement.M
STOR: 'Command' = 'STOR', FEATCode.base, 'Store', CommandType.S, ConformanceRequirement.M
STOU: 'Command' = 'STOU', FEATCode.base, 'Store Unique', CommandType.A, ConformanceRequirement.O
STRU: 'Command' = 'STRU', FEATCode.base, 'File Structure', CommandType.P, ConformanceRequirement.M
SYST: 'Command' = 'SYST', FEATCode.base, 'System', CommandType.S, ConformanceRequirement.O
TYPE: 'Command' = 'TYPE', FEATCode.base, 'Representation Type', CommandType.P, ConformanceRequirement.M
USER: 'Command' = 'USER', FEATCode.base, 'User Name', CommandType.A, ConformanceRequirement.M
XCUP: 'Command' = 'XCUP', FEATCode.hist, None, CommandType.S, ConformanceRequirement.H
XCWD: 'Command' = 'XCWD', FEATCode.hist, None, CommandType.S, ConformanceRequirement.H
XMKD: 'Command' = 'XMKD', FEATCode.hist, None, CommandType.S, ConformanceRequirement.H
XPWD: 'Command' = 'XPWD', FEATCode.hist, None, CommandType.S, ConformanceRequirement.H
XRMD: 'Command' = 'XRMD', FEATCode.hist, None, CommandType.S, ConformanceRequirement.H
TVFS: 'Command' = 'TVFS', FEATCode('TVFS'), 'Trivial Virtual File Store', CommandType.P, ConformanceRequirement.O
```

--------------------------------

### Read Interface Statistics Start Time Option

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/misc/pcapng.html

Validates the Interface Statistics Block context and constructs a Data_ISB_StartTimeOption.

```python
    def _read_option_isb_starttime(self, schema: 'Schema_ISB_StartTimeOption', *,
                                   options: 'Option') -> 'Data_ISB_StartTimeOption':
        """Read PCAP-NG ``isb_starttime`` option.

        Args:
            schema: Parsed option schema.
            options: Parsed PCAP-NG options.

        Returns:
            Constructed option data.

        """
        if self._type != Enum_BlockType.Interface_Statistics_Block:
            raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be in Interface Statistics Block, '
                                f'but found in {self._type} block.')
        if self._opt[schema.type] > 0:
            raise ProtocolError(f'PCAP-NG: [isb_starttime] option must be only one, '
                                f'but {self._opt[schema.type] + 1} found.')
        if schema.length != 8:
            raise ProtocolError(f'PCAP-NG: [isb_starttime] invalid length (expected 8, got {schema.length})')

        timestamp, timestamp_epoch = self._read_timestamp(schema.timestamp_high, schema.timestamp_low,
                                                          interface_id=self._isb_interface_id)

        option = Data_ISB_StartTimeOption(
            type=schema.type,
            length=schema.length,
            timestamp=timestamp,
            timestamp_epoch=timestamp_epoch,
        )
        return option
```

--------------------------------

### Quick-Start Request Option Structure

Source: https://jarryshaw.github.io/PyPCAPKit/_modules/pcapkit/protocols/internet/ipv6_opts.html

Visual representation of the Quick-Start Request option format as defined in RFC 4782.

```text
0                   1                   2                   3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|   Option      |  Length=6     | Func. | Rate  |   QS TTL      |
|               |               | 0000  |Request|               |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                        QS Nonce                           | R |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
```
Tokens:
Raw