| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -15,6 +15,8 @@ cdef bint TYPE_CHECKING | |||
| 15 | 15 | cdef cython.uint _MAX_MSG_ABSOLUTE | |
| 16 | 16 | cdef cython.uint _DUPLICATE_PACKET_SUPPRESSION_INTERVAL | |
| 17 | 17 | cdef cython.uint _RECENT_PACKETS_MAX | |
| 18 | + cdef cython.uint _MAX_DEFERRED_ADDRS | ||
| 19 | + cdef cython.uint _MAX_DEFERRED_PER_ADDR | ||
| 18 | 20 | ||
| 19 | 21 | ||
| 20 | 22 | cdef class AsyncListener: | |
@@ -41,6 +43,8 @@ cdef class AsyncListener: | |||
| 41 | 43 | ||
| 42 | 44 | cdef _cancel_any_timers_for_addr(self, object addr) | |
| 43 | 45 | ||
| 46 | + cdef _evict_oldest_deferred(self) | ||
| 47 | + | ||
| 44 | 48 | @cython.locals(deadline=object, fire_at=double) | |
| 45 | 49 | cdef double _compute_deferred_fire_at(self, object addr, double now, double delay) | |
| 46 | 50 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -32,7 +32,13 @@ | |||
| 32 | 32 | from ._protocol.incoming import DNSIncoming | |
| 33 | 33 | from ._transport import _WrappedTransport, make_wrapped_transport | |
| 34 | 34 | from ._utils.time import current_time_millis, millis_to_seconds | |
| 35 | - from .const import _DUPLICATE_PACKET_SUPPRESSION_INTERVAL, _MAX_MSG_ABSOLUTE, _RECENT_PACKETS_MAX | ||
| 35 | + from .const import ( | ||
| 36 | + _DUPLICATE_PACKET_SUPPRESSION_INTERVAL, | ||
| 37 | + _MAX_DEFERRED_ADDRS, | ||
| 38 | + _MAX_DEFERRED_PER_ADDR, | ||
| 39 | + _MAX_MSG_ABSOLUTE, | ||
| 40 | + _RECENT_PACKETS_MAX, | ||
| 41 | + ) | ||
| 36 | 42 | ||
| 37 | 43 | if TYPE_CHECKING: | |
| 38 | 44 | from ._core import Zeroconf | |
@@ -240,7 +246,17 @@ def handle_query_or_defer( | |||
| 240 | 246 | self._respond_query(msg, addr, port, transport, v6_flow_scope) | |
| 241 | 247 | return | |
| 242 | 248 | ||
| 249 | + if addr not in self._deferred and len(self._deferred) >= _MAX_DEFERRED_ADDRS: | ||
| 250 | + # Bound total deferred addrs so a spoofed-source flood | ||
| 251 | + # cannot keep adding distinct entries; evict the oldest | ||
| 252 | + # (insertion-order) entry and discard its in-flight queue. | ||
| 253 | + self._evict_oldest_deferred() | ||
| 254 | + | ||
| 243 | 255 | deferred = self._deferred.setdefault(addr, []) | |
| 256 | + if len(deferred) >= _MAX_DEFERRED_PER_ADDR: | ||
| 257 | + # Bound per-addr queue length; further fragments from the | ||
| 258 | + # same source are dropped until the timer flushes. | ||
| 259 | + return | ||
| 244 | 260 | # If we get the same packet we ignore it | |
| 245 | 261 | for incoming in reversed(deferred): | |
| 246 | 262 | if incoming.data == msg.data: | |
@@ -293,6 +309,21 @@ def _cancel_any_timers_for_addr(self, addr: _str) -> None: | |||
| 293 | 309 | if addr in self._timers: | |
| 294 | 310 | self._timers.pop(addr).cancel() | |
| 295 | 311 | ||
| 312 | + def _evict_oldest_deferred(self) -> None: | ||
| 313 | + """Discard the oldest deferred addr's reassembly state. | ||
| 314 | + | ||
| 315 | + Used when ``_MAX_DEFERRED_ADDRS`` would be exceeded; the | ||
| 316 | + evicted addr's queue and timer are dropped without firing, so | ||
| 317 | + the bound holds even when an attacker rotates source IPs. | ||
| 318 | + Eviction is FIFO (oldest by first-seen, via dict insertion | ||
| 319 | + order) rather than LRU so an active flooder cannot pin its | ||
| 320 | + slots by re-sending into the same addr. | ||
| 321 | + """ | ||
| 322 | + oldest_addr = next(iter(self._deferred)) | ||
| 323 | + self._cancel_any_timers_for_addr(oldest_addr) | ||
| 324 | + self._deferred_deadlines.pop(oldest_addr, None) | ||
| 325 | + del self._deferred[oldest_addr] | ||
| 326 | + | ||
| 296 | 327 | def _respond_query( | |
| 297 | 328 | self, | |
| 298 | 329 | msg: DNSIncoming | None, | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -77,6 +77,20 @@ | |||
| 77 | 77 | # flooding distinct questions (RFC 6762 §7.3, defense-in-depth). | |
| 78 | 78 | _MAX_QUESTION_HISTORY_ENTRIES = 10000 | |
| 79 | 79 | ||
| 80 | + # Per-addr cap on the number of truncated (TC-bit) packets retained for | ||
| 81 | + # RFC 6762 §18.5 reassembly. The spec anticipates only a handful of | ||
| 82 | + # segments per truncated query; 16 is well above legitimate need and | ||
| 83 | + # keeps the per-arrival dedup scan a constant-time cost under a flood. | ||
| 84 | + _MAX_DEFERRED_PER_ADDR = 16 | ||
| 85 | + | ||
| 86 | + # Per-listener cap on the number of distinct addrs with in-flight | ||
| 87 | + # TC-deferral state. Each entry can hold up to _MAX_DEFERRED_PER_ADDR | ||
| 88 | + # packets of up to _MAX_MSG_ABSOLUTE bytes; 512 leaves headroom for a | ||
| 89 | + # legitimate burst (LAN-wide power-resume / boot storm where many | ||
| 90 | + # devices announce at once) while bounding worst-case memory at | ||
| 91 | + # ~72 MB even when a peer floods with spoofed source IPs. | ||
| 92 | + _MAX_DEFERRED_ADDRS = 512 | ||
| 93 | + | ||
| 80 | 94 | _DNS_PACKET_HEADER_LEN = 12 | |
| 81 | 95 | ||
| 82 | 96 | _MAX_MSG_TYPICAL = 1460 # unused | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -19,7 +19,7 @@ | |||
| 19 | 19 | import pytest | |
| 20 | 20 | ||
| 21 | 21 | import zeroconf as r | |
| 22 | - from zeroconf import NotRunningException, Zeroconf, const, current_time_millis | ||
| 22 | + from zeroconf import NotRunningException, Zeroconf, _listener, const, current_time_millis | ||
| 23 | 23 | from zeroconf._listener import _TC_DELAY_RANDOM_INTERVAL, AsyncListener, _WrappedTransport | |
| 24 | 24 | from zeroconf._protocol.incoming import DNSIncoming | |
| 25 | 25 | from zeroconf.asyncio import AsyncZeroconf | |
@@ -794,6 +794,101 @@ def test_tc_bit_defer_window_is_bounded(): | |||
| 794 | 794 | zc.close() | |
| 795 | 795 | ||
| 796 | 796 | ||
| 797 | + def _make_distinct_tc_packets(count: int, name_prefix: str = "q") -> list[bytes]: | ||
| 798 | + """Generate ``count`` byte-distinct TC-flagged query packets for flood inputs.""" | ||
| 799 | + packets = [] | ||
| 800 | + for i in range(count): | ||
| 801 | + out = r.DNSOutgoing(const._FLAGS_QR_QUERY | const._FLAGS_TC) | ||
| 802 | + out.add_question(r.DNSQuestion(f"{name_prefix}{i}._tcp.local.", const._TYPE_PTR, const._CLASS_IN)) | ||
| 803 | + packets.append(out.packets()[0]) | ||
| 804 | + return packets | ||
| 805 | + | ||
| 806 | + | ||
| 807 | + def _synthetic_source_ip(i: int) -> str: | ||
| 808 | + """Distinct synthetic source IPs from the documentation ranges.""" | ||
| 809 | + if i < 256: | ||
| 810 | + return f"203.0.113.{i}" | ||
| 811 | + if i < 512: | ||
| 812 | + return f"198.51.100.{i - 256}" | ||
| 813 | + return f"192.0.2.{i - 512}" | ||
| 814 | + | ||
| 815 | + | ||
| 816 | + def test_tc_bit_per_addr_queue_is_bounded(quick_timing: None) -> None: | ||
| 817 | + """Per-addr deferred queue must not grow past ``_MAX_DEFERRED_PER_ADDR``.""" | ||
| 818 | + zc = Zeroconf(interfaces=["127.0.0.1"]) | ||
| 819 | + _wait_for_start(zc) | ||
| 820 | + protocol = zc.engine.protocols[0] | ||
| 821 | + source_ip = "203.0.113.21" | ||
| 822 | + | ||
| 823 | + extra = 4 | ||
| 824 | + packets = _make_distinct_tc_packets(const._MAX_DEFERRED_PER_ADDR + extra) | ||
| 825 | + | ||
| 826 | + # Push the reassembly timer well past any possible test runtime | ||
| 827 | + # so the bound under test is the only thing that can drop entries. | ||
| 828 | + with patch.object(_listener, "_TC_DELAY_RANDOM_INTERVAL", (60_000, 60_001)): | ||
| 829 | + for raw in packets: | ||
| 830 | + threadsafe_query(zc, protocol, r.DNSIncoming(raw), source_ip, const._MDNS_PORT, Mock(), ()) | ||
| 831 | + | ||
| 832 | + assert len(protocol._deferred[source_ip]) == const._MAX_DEFERRED_PER_ADDR | ||
| 833 | + # Last ``extra`` packets must have been dropped, not displaced; the | ||
| 834 | + # earlier ``_MAX_DEFERRED_PER_ADDR`` entries are the ones retained. | ||
| 835 | + retained = [incoming.data for incoming in protocol._deferred[source_ip]] | ||
| 836 | + assert retained == packets[: const._MAX_DEFERRED_PER_ADDR] | ||
| 837 | + | ||
| 838 | + zc.close() | ||
| 839 | + | ||
| 840 | + | ||
| 841 | + def test_tc_bit_total_addrs_is_bounded(quick_timing: None) -> None: | ||
| 842 | + """Distinct addrs with deferred state must not exceed ``_MAX_DEFERRED_ADDRS``.""" | ||
| 843 | + zc = Zeroconf(interfaces=["127.0.0.1"]) | ||
| 844 | + _wait_for_start(zc) | ||
| 845 | + protocol = zc.engine.protocols[0] | ||
| 846 | + | ||
| 847 | + raw = _make_distinct_tc_packets(1)[0] | ||
| 848 | + extra = 4 | ||
| 849 | + addrs = [_synthetic_source_ip(i) for i in range(const._MAX_DEFERRED_ADDRS + extra)] | ||
| 850 | + | ||
| 851 | + # Push the reassembly timer well past any possible test runtime | ||
| 852 | + # so the bound under test is the only thing that can drop entries; | ||
| 853 | + # without this, PyPy / slow runners can fire timers between the | ||
| 854 | + # last enqueue and the assertion. | ||
| 855 | + with patch.object(_listener, "_TC_DELAY_RANDOM_INTERVAL", (60_000, 60_001)): | ||
| 856 | + for source_ip in addrs: | ||
| 857 | + threadsafe_query(zc, protocol, r.DNSIncoming(raw), source_ip, const._MDNS_PORT, Mock(), ()) | ||
| 858 | + | ||
| 859 | + assert len(protocol._deferred) == const._MAX_DEFERRED_ADDRS | ||
| 860 | + assert len(protocol._timers) == const._MAX_DEFERRED_ADDRS | ||
| 861 | + | ||
| 862 | + zc.close() | ||
| 863 | + | ||
| 864 | + | ||
| 865 | + def test_tc_bit_eviction_drops_oldest_addr(quick_timing: None) -> None: | ||
| 866 | + """Adding a new addr at capacity drops the oldest insertion (FIFO).""" | ||
| 867 | + zc = Zeroconf(interfaces=["127.0.0.1"]) | ||
| 868 | + _wait_for_start(zc) | ||
| 869 | + protocol = zc.engine.protocols[0] | ||
| 870 | + | ||
| 871 | + raw = _make_distinct_tc_packets(1)[0] | ||
| 872 | + fillers = [_synthetic_source_ip(i) for i in range(const._MAX_DEFERRED_ADDRS)] | ||
| 873 | + new_addr = _synthetic_source_ip(const._MAX_DEFERRED_ADDRS) | ||
| 874 | + oldest = fillers[0] | ||
| 875 | + | ||
| 876 | + with patch.object(_listener, "_TC_DELAY_RANDOM_INTERVAL", (60_000, 60_001)): | ||
| 877 | + for source_ip in fillers: | ||
| 878 | + threadsafe_query(zc, protocol, r.DNSIncoming(raw), source_ip, const._MDNS_PORT, Mock(), ()) | ||
| 879 | + assert len(protocol._deferred) == const._MAX_DEFERRED_ADDRS | ||
| 880 | + assert oldest in protocol._deferred | ||
| 881 | + | ||
| 882 | + # One more distinct addr must evict the oldest insertion-order entry. | ||
| 883 | + threadsafe_query(zc, protocol, r.DNSIncoming(raw), new_addr, const._MDNS_PORT, Mock(), ()) | ||
| 884 | + assert oldest not in protocol._deferred | ||
| 885 | + assert oldest not in protocol._timers | ||
| 886 | + assert new_addr in protocol._deferred | ||
| 887 | + assert len(protocol._deferred) == const._MAX_DEFERRED_ADDRS | ||
| 888 | + | ||
| 889 | + zc.close() | ||
| 890 | + | ||
| 891 | + | ||
| 797 | 892 | @pytest.mark.asyncio | |
| 798 | 893 | async def test_open_close_twice_from_async() -> None: | |
| 799 | 894 | """Test we can close twice from a coroutine when using Zeroconf. | |
| Back | FazBrowse Home | New Git URL |
0 commit comments