Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion pcapkit/foundation/engines/pcapng.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from typing import TYPE_CHECKING, cast

from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType
from pcapkit.corekit.fields.strings import pad_length
from pcapkit.corekit.infoclass import Info, info_final
from pcapkit.corekit.io import SeekableReader
from pcapkit.dumpkit.pcap import PCAPIO
Expand Down Expand Up @@ -458,7 +459,7 @@ def _bound_read_ahead(self) -> 'None':
size = length
if block_type == Enum_BlockType.Decryption_Secrets_Block:
secrets = int.from_bytes(self._peek(16)[12:16], byteorder)
size = max(size, BLOCK_MINIMUM[block_type] + secrets + -secrets % 4)
size = max(size, BLOCK_MINIMUM[block_type] + secrets + pad_length(secrets, 4))
# NOTE: No buffer holds more than :data:`sys.maxsize` octets, so a block
# longer than that is read ahead only so far; the reader moves past the
# rest itself (#1975).
Expand Down
3 changes: 2 additions & 1 deletion pcapkit/foundation/engines/scapy.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import struct
from typing import TYPE_CHECKING, cast

from pcapkit.corekit.fields.strings import pad_length
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.foundation.reassembly import ReassemblyManager
from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN
Expand Down Expand Up @@ -106,7 +107,7 @@ def _read_tsoffset(options: 'bytes', endian: 'str') -> 'int':
break
if code == 14 and length == 8 and len(options) >= 12:
return cast('int', struct.unpack(f'{endian}q', options[4:12])[0])
options = options[4 + length + (-length % 4):]
options = options[4 + length + pad_length(length, 4):]
return 0


Expand Down
27 changes: 9 additions & 18 deletions pcapkit/protocols/link/arp.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,6 @@

"""
import ipaddress
import re
from typing import TYPE_CHECKING, cast

from pcapkit.const.arp.hardware import Hardware as Enum_Hardware
Expand Down Expand Up @@ -72,9 +71,6 @@

__all__ = ['ARP']

# Ethernet address pattern, of any number of octets (``hlen`` is not fixed at 6)
PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}(?:[:-][0-9a-f]{2})*)?')


class ARP(Link[Data_ARP, Schema_ARP],
schema=Schema_ARP, data=Data_ARP):
Expand Down Expand Up @@ -374,10 +370,8 @@ def _read_addr_resolve(self, addr: 'bytes', htype: 'int') -> 'str':

"""
if htype == Enum_Hardware.Ethernet: # Ethernet
_addr = addr.hex(':')
else:
_addr = addr.hex()
return _addr
return self._read_mac_addr(addr)
return addr.hex()

def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> 'bytes':
"""Resolve hardware address according to protocol.
Expand All @@ -389,8 +383,9 @@ def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> '
Returns:
Hardware address packed as raw octets. If ``htype`` is ``1``, i.e.
MAC address, the ``:``- or ``-``-separated *hex* encoded MAC address
is validated and packed, one octet per group (an empty string packs
as no octets); otherwise, a :class:`str` is
is packed by :meth:`~pcapkit.protocols.protocol.Protocol._make_mac_addr`,
one octet per group, of any number since ``hlen`` is not fixed at 6
(an empty string packs as no octets); otherwise, a :class:`str` is
taken as the *hex* encoded address that :meth:`_read_addr_resolve`
returns, and :class:`bytes`/:class:`bytearray` are returned as
:class:`bytes`.
Expand All @@ -409,20 +404,16 @@ def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> '
:meth:`_make_proto_resolve` gives.

"""
name = 'MAC address' if htype == Enum_Hardware.Ethernet else 'hardware address'
if htype == Enum_Hardware.Ethernet:
return self._make_mac_addr(addr, size=None, alias=self.__class__.__name__)
if not isinstance(addr, (str, bytes, bytearray)):
raise ProtocolError(f'{self.__class__.__name__}: invalid {name}: must be a str, '
raise ProtocolError(f'{self.__class__.__name__}: invalid hardware address: must be a str, '
f'bytes or bytearray, not {addr!r}')
if htype == Enum_Hardware.Ethernet:
_addr = addr.encode() if isinstance(addr, str) else addr
if PAT_MAC_ADDR.fullmatch(_addr) is not None:
return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode())
raise ProtocolError(f'{self.__class__.__name__}: invalid {name}: {addr!r}')
if isinstance(addr, str):
try:
return bytes.fromhex(addr.replace(':', '').replace('-', ''))
except ValueError:
raise ProtocolError(f'{self.__class__.__name__}: invalid {name}: '
raise ProtocolError(f'{self.__class__.__name__}: invalid hardware address: '
f'{addr!r}') from None
return bytes(addr)

Expand Down
Loading
Loading