### 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__ `. Returns: Payload of :attr:`self.__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 ``` -------------------------------- ### 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 `. """ #: 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 `). _protocol (Union[str, Protocol, Type[Protocol]]): Parse packet until ``_protocol`` (:attr:`self._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 | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ ```