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
16 changes: 16 additions & 0 deletions CHANGELOG.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -8,6 +8,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]

### Changed
- **The decoder no longer re-validates the addresses it just
generated.** Every header's `__post_init__` ran the address
validators, including for instances built by `decode()` — so a MAC
mechanically rendered from six bytes by `bytes_to_mac()` was
immediately matched against `mac_regex` to confirm what the
conversion had already guaranteed. `Ethernet`, `ARP` and `IPv4` now
build their decoded instance directly (`object.__new__` plus
`object.__setattr__` per field), skipping `__init__` and
`__post_init__` on that path only. Measured here: `Ethernet.decode`
2330 → 830 ns (2.8x), `IPv4.decode` 4936 → 2596 ns (1.9x), a corpus
walk 76,500 → 113,200 frames/sec (1.48x). **Strictness on
construction is unchanged** — every public constructor still
validates and still raises, which is now asserted alongside a test
that the decode path runs no regex at all. The `__post_init__` checks
bypassed this way are ones `decode()` establishes itself (documented
at each site and in `_base.py`). No API change.
- **Protocol dispatch is a table lookup, not a table construction.**
`_ethertype_class()` and `_ip_protocol_class()` re-ran their deferred
imports and rebuilt a `dict` literal on *every* call — once per layer
Expand Down
18 changes: 18 additions & 0 deletions src/netprotocols/_base.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,24 @@

``memoryview`` input is accepted and used as a decode-time transient
only; no view is ever stored on an instance.

Decode-path construction
------------------------
Constructing a header from field values validates them: an address that
is not a valid MAC or IPv4 string raises, and that strictness is part
of the public contract. It is pure overhead on the decode path, though,
because ``decode()`` *generates* those strings itself, from raw bytes,
via ``bytes_to_mac()`` / ``bytes_to_ipv4()`` — a regex can only confirm
what the conversion already guarantees.

The three headers that carry addresses (``Ethernet``, ``ARP``,
``IPv4``) therefore build their decoded instance directly, with
``object.__new__`` plus ``object.__setattr__`` per field, skipping
``__init__`` and ``__post_init__``. This is an internal shortcut on a
path whose inputs are known-good, not a relaxation of the contract:
every public constructor still validates, and any ``__post_init__``
check bypassed this way is one ``decode()`` has already established
itself (documented at each site).
"""

from __future__ import annotations
Expand Down
27 changes: 16 additions & 11 deletions src/netprotocols/layer2/arp.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -61,17 +61,22 @@ def decode(cls, data: bytes | memoryview) -> Self:
htype, ptype, hlen, plen, oper, sha, spa, tha, tpa = cls._unpack_fixed(
data
)
return cls(
htype=htype,
ptype=ptype,
hlen=hlen,
plen=plen,
oper=oper,
sha=bytes_to_mac(sha),
spa=bytes_to_ipv4(spa),
tha=bytes_to_mac(tha),
tpa=bytes_to_ipv4(tpa),
)
# The addresses below are generated here, from raw bytes, by
# bytes_to_mac()/bytes_to_ipv4() — they cannot fail the
# validators the constructor runs, so this builds the instance
# directly instead (see "Decode-path construction" in _base.py).
header = object.__new__(cls)
set_field = object.__setattr__
set_field(header, "htype", htype)
set_field(header, "ptype", ptype)
set_field(header, "hlen", hlen)
set_field(header, "plen", plen)
set_field(header, "oper", oper)
set_field(header, "sha", bytes_to_mac(sha))
set_field(header, "spa", bytes_to_ipv4(spa))
set_field(header, "tha", bytes_to_mac(tha))
set_field(header, "tpa", bytes_to_ipv4(tpa))
return header

def __bytes__(self) -> bytes:
return self._struct.pack(
Expand Down
13 changes: 10 additions & 3 deletions src/netprotocols/layer2/ethernet.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -88,9 +88,16 @@ def __post_init__(self) -> None:
@classmethod
def decode(cls, data: bytes | memoryview) -> Self:
dst, src, ethertype = cls._unpack_fixed(data)
return cls(
dst=bytes_to_mac(dst), src=bytes_to_mac(src), ethertype=ethertype
)
# The addresses below are generated here, from raw bytes, by
# bytes_to_mac()/bytes_to_ipv4() — they cannot fail the
# validators the constructor runs, so this builds the instance
# directly instead (see "Decode-path construction" in _base.py).
header = object.__new__(cls)
set_field = object.__setattr__
set_field(header, "dst", bytes_to_mac(dst))
set_field(header, "src", bytes_to_mac(src))
set_field(header, "ethertype", ethertype)
return header

def __bytes__(self) -> bytes:
return self._struct.pack(
Expand Down
38 changes: 22 additions & 16 deletions src/netprotocols/layer3/ip.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -239,22 +239,28 @@ def decode(cls, data: bytes | memoryview) -> Self:
f"IPv4 header declares {ihl * 4} bytes, buffer holds "
f"{len(data)}"
)
return cls(
version=ver_ihl >> 4,
ihl=ihl,
dscp=dscp_ecn >> 2,
ecn=dscp_ecn & 0b11,
total_length=total_length,
identification=identification,
flags=flags_frag >> 13,
fragment_offset=flags_frag & 0x1FFF,
ttl=ttl,
protocol=protocol,
checksum=checksum,
src=bytes_to_ipv4(src),
dst=bytes_to_ipv4(dst),
options=bytes(data[cls._struct.size : ihl * 4]),
)
# As in Ethernet/ARP, the addresses are generated here and
# cannot fail validation. The constructor's other checks are
# already established above: the IHL is 4 bits (so <= 15), it
# was rejected below 5, and the options are sliced to exactly
# ihl * 4 bytes after the buffer was confirmed to hold them.
header = object.__new__(cls)
set_field = object.__setattr__
set_field(header, "version", ver_ihl >> 4)
set_field(header, "ihl", ihl)
set_field(header, "dscp", dscp_ecn >> 2)
set_field(header, "ecn", dscp_ecn & 0b11)
set_field(header, "total_length", total_length)
set_field(header, "identification", identification)
set_field(header, "flags", flags_frag >> 13)
set_field(header, "fragment_offset", flags_frag & 0x1FFF)
set_field(header, "ttl", ttl)
set_field(header, "protocol", protocol)
set_field(header, "checksum", checksum)
set_field(header, "src", bytes_to_ipv4(src))
set_field(header, "dst", bytes_to_ipv4(dst))
set_field(header, "options", bytes(data[cls._struct.size : ihl * 4]))
return header

def __bytes__(self) -> bytes:
return (
Expand Down
114 changes: 114 additions & 0 deletions tests/test_contract.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -9,6 +9,8 @@
UDP,
Ethernet,
ICMPv4,
InvalidIPv4AddressError,
InvalidMACAddressError,
IPv4,
IPv6,
ProtocolError,
Expand DownExpand Up@@ -106,3 +108,115 @@ def test_unknown_ip_protocol_ends_chain(self, raw_ipv4_header):
b"\x45" + raw_ipv4_header[1:9] + b"\xfd" + raw_ipv4_header[10:]
)
assert IPv4.decode(unknown).next_protocol() is None


class TestDecodePathValidation:
"""The decoder does not re-validate strings it generated itself,
but every public constructor still does (#84)."""

def spy_on(self, module, name):
"""Replace a compiled regex with a spy that fails if used."""
from unittest import mock

spy = mock.Mock()
spy.match.side_effect = AssertionError(
f"{name} matched on the decode path"
)
return mock.patch.object(module, name, spy)

def test_ethernet_decode_runs_no_mac_regex(self, raw_eth_header):
from netprotocols.utils import mac

with self.spy_on(mac, "mac_regex"):
eth = Ethernet.decode(raw_eth_header)
assert eth.dst == "ff:ff:ff:ff:ff:ff"
assert bytes(eth) == raw_eth_header

def test_arp_decode_runs_no_regex(self, raw_arp_header):
from netprotocols.utils import ipv4, mac

with self.spy_on(mac, "mac_regex"), self.spy_on(ipv4, "ipv4_regex"):
arp = ARP.decode(raw_arp_header)
assert arp.sha == "00:07:0d:af:f4:54"
assert arp.tpa == "24.166.173.159"
assert bytes(arp) == raw_arp_header

def test_ipv4_decode_runs_no_ipv4_regex(self, raw_ipv4_header):
from netprotocols.utils import ipv4

with self.spy_on(ipv4, "ipv4_regex"):
ip = IPv4.decode(raw_ipv4_header)
assert ip.src == "192.168.1.96"
assert bytes(ip) == raw_ipv4_header

def test_construction_still_validates(self):
"""The strictness the decode path skips is intact for callers."""
with pytest.raises(InvalidMACAddressError):
Ethernet(dst="nonsense", src="00:07:0d:af:f4:54", ethertype=0x0800)
with pytest.raises(InvalidMACAddressError):
ARP(
htype=1,
ptype=0x0800,
hlen=6,
plen=4,
oper=1,
sha="not-a-mac",
spa="192.0.2.1",
tha="00:00:00:00:00:00",
tpa="192.0.2.2",
)
with pytest.raises(InvalidIPv4AddressError):
IPv4(
version=4,
ihl=5,
dscp=0,
ecn=0,
total_length=20,
identification=1,
flags=2,
fragment_offset=0,
ttl=64,
protocol=6,
checksum=0,
src="999.1.1.1",
dst="192.0.2.2",
)

def test_decoded_instances_equal_constructed_ones(
self, raw_eth_header, raw_arp_header, raw_ipv4_header
):
"""Equality compares every field, so a field the bypass forgot
to set would raise AttributeError here rather than lurk."""
eth = Ethernet.decode(raw_eth_header)
assert eth == Ethernet(
dst="ff:ff:ff:ff:ff:ff", src="00:07:0d:af:f4:54", ethertype=0x0806
)
arp = ARP.decode(raw_arp_header)
assert arp == ARP(
htype=1,
ptype=0x0800,
hlen=6,
plen=4,
oper=1,
sha="00:07:0d:af:f4:54",
spa="24.166.172.1",
tha="00:00:00:00:00:00",
tpa="24.166.173.159",
)
ip = IPv4.decode(raw_ipv4_header)
assert ip == IPv4(
version=4,
ihl=5,
dscp=0,
ecn=0,
total_length=40,
identification=0xEC6C,
flags=2,
fragment_offset=0,
ttl=64,
protocol=6,
checksum=0x2B51,
src="192.168.1.96",
dst="192.168.1.254",
)
assert hash(eth) and hash(arp) and hash(ip)
Loading