refactor: share _mark_seen helper between _logger and _protocol.incoming

bdraco · bdraco · commit 85ac2ae21d5a · 2026-05-17T19:11:40.000-07:00
Lift the bound-check + eviction logic out of the four QuietLogger
classmethods and the parser-side `_log_exception_debug` and into a
single `_mark_seen(seen, key)` in `_logger.py`. Both call sites still
keep their own `_seen_logs` set (no behavior change vs the previous
commit — there are still two independent dedup pools, each capped at
`_MAX_SEEN_LOGS`); only the eviction policy is now defined in one
place. `_MAX_SEEN_LOGS` likewise moves to `_logger.py` only.
diff --git a/src/zeroconf/_logger.py b/src/zeroconf/_logger.py
@@ -25,7 +25,7 @@
 
 import logging
 import sys
-from typing import Any, ClassVar
+from typing import Any
 
 log = logging.getLogger(__name__.split(".", maxsplit=1)[0])
 log.addHandler(logging.NullHandler())
@@ -40,41 +40,41 @@ def set_logger_level_if_unset() -> None:
 
 
 _MAX_SEEN_LOGS = 256
+_seen_logs: set[str] = set()
 
 
-class QuietLogger:
-    _seen_logs: ClassVar[set[str]] = set()
+def _mark_seen(seen: set[str], key: str) -> bool:
+    """Record ``key`` in ``seen`` and return True if it was newly added.
 
-    @classmethod
-    def _mark_seen(cls, key: str) -> bool:
-        """Record ``key`` and return True if it was newly added."""
-        if key in cls._seen_logs:
-            return False
-        # Keys can carry caller-supplied fields (peer addresses, packet
-        # offsets); clear when full so a malicious peer can't grow the
-        # set without bound.
-        if len(cls._seen_logs) >= _MAX_SEEN_LOGS:
-            cls._seen_logs.clear()
-        cls._seen_logs.add(key)
-        return True
+    Bounds the set so callers passing attacker-influenced keys (peer
+    addresses, packet offsets) cannot grow it without bound.
+    """
+    if key in seen:
+        return False
+    if len(seen) >= _MAX_SEEN_LOGS:
+        seen.clear()
+    seen.add(key)
+    return True
 
+
+class QuietLogger:
     @classmethod
     def log_exception_warning(cls, *logger_data: Any) -> None:
-        first_time = cls._mark_seen(str(sys.exc_info()[1]))
+        first_time = _mark_seen(_seen_logs, str(sys.exc_info()[1]))
         logger = log.warning if first_time else log.debug
         logger(*(logger_data or ["Exception occurred"]), exc_info=True)
 
     @classmethod
     def log_exception_debug(cls, *logger_data: Any) -> None:
-        first_time = cls._mark_seen(str(sys.exc_info()[1]))
+        first_time = _mark_seen(_seen_logs, str(sys.exc_info()[1]))
         log.debug(*(logger_data or ["Exception occurred"]), exc_info=first_time)
 
     @classmethod
     def log_warning_once(cls, *args: Any) -> None:
-        logger = log.warning if cls._mark_seen(args[0]) else log.debug
+        logger = log.warning if _mark_seen(_seen_logs, args[0]) else log.debug
         logger(*args)
 
     @classmethod
     def log_exception_once(cls, exc: Exception, *args: Any) -> None:
-        logger = log.warning if cls._mark_seen(args[0]) else log.debug
+        logger = log.warning if _mark_seen(_seen_logs, args[0]) else log.debug
         logger(*args, exc_info=exc)
diff --git a/src/zeroconf/_protocol/incoming.py b/src/zeroconf/_protocol/incoming.py
@@ -37,7 +37,7 @@
     DNSText,
 )
 from .._exceptions import IncomingDecodeError
-from .._logger import log
+from .._logger import _mark_seen, log
 from .._utils.time import current_time_millis
 from ..const import (
     _FLAGS_QR_MASK,
@@ -63,7 +63,6 @@
 DECODE_EXCEPTIONS = (IndexError, struct.error, IncomingDecodeError)
 
 
-_MAX_SEEN_LOGS = 256
 _seen_logs: set[str] = set()
 _str = str
 _int = int
@@ -183,16 +182,7 @@ def _initial_parse(self) -> None:
 
     @classmethod
     def _log_exception_debug(cls, *logger_data: Any) -> None:
-        log_exc_info = False
-        exc_str = str(sys.exc_info()[1])
-        if exc_str not in _seen_logs:
-            # The dedup key embeds attacker-controlled fields (peer IP/port,
-            # byte offsets); clear when full so a malicious peer can't grow
-            # the set without bound.
-            if len(_seen_logs) >= _MAX_SEEN_LOGS:
-                _seen_logs.clear()
-            _seen_logs.add(exc_str)
-            log_exc_info = True
+        log_exc_info = _mark_seen(_seen_logs, str(sys.exc_info()[1]))
         log.debug(*(logger_data or ["Exception occurred"]), exc_info=log_exc_info)
 
     def answers(self) -> list[DNSRecord]:
diff --git a/tests/test_logger.py b/tests/test_logger.py
@@ -5,6 +5,7 @@
 import logging
 from unittest.mock import call, patch
 
+from zeroconf import _logger
 from zeroconf._logger import _MAX_SEEN_LOGS, QuietLogger, set_logger_level_if_unset
 
 
@@ -25,7 +26,7 @@ def test_loading_logger():
 
 def test_log_warning_once():
     """Test we only log with warning level once."""
-    QuietLogger._seen_logs = set()
+    _logger._seen_logs.clear()
     quiet_logger = QuietLogger()
     with (
         patch("zeroconf._logger.log.warning") as mock_log_warning,
@@ -48,7 +49,7 @@ def test_log_warning_once():
 
 def test_log_exception_warning():
     """Test we only log with warning level once."""
-    QuietLogger._seen_logs = set()
+    _logger._seen_logs.clear()
     quiet_logger = QuietLogger()
     with (
         patch("zeroconf._logger.log.warning") as mock_log_warning,
@@ -71,7 +72,7 @@ def test_log_exception_warning():
 
 def test_llog_exception_debug():
     """Test we only log with a trace once."""
-    QuietLogger._seen_logs = set()
+    _logger._seen_logs.clear()
     quiet_logger = QuietLogger()
     with patch("zeroconf._logger.log.debug") as mock_log_debug:
         quiet_logger.log_exception_debug("the exception")
@@ -86,16 +87,16 @@ def test_llog_exception_debug():
 
 def test_seen_logs_is_bounded():
     """Distinct keys must not grow ``_seen_logs`` without bound."""
-    QuietLogger._seen_logs = set()
+    _logger._seen_logs.clear()
     with patch("zeroconf._logger.log.warning"), patch("zeroconf._logger.log.debug"):
         for i in range(_MAX_SEEN_LOGS + 5):
             QuietLogger.log_warning_once(f"warning-{i}")
-    assert len(QuietLogger._seen_logs) <= _MAX_SEEN_LOGS
+    assert len(_logger._seen_logs) <= _MAX_SEEN_LOGS
 
 
 def test_log_exception_once():
     """Test we only log with warning level once."""
-    QuietLogger._seen_logs = set()
+    _logger._seen_logs.clear()
     quiet_logger = QuietLogger()
     exc = Exception()
     with (
diff --git a/tests/test_protocol.py b/tests/test_protocol.py
@@ -14,6 +14,7 @@
 
 import zeroconf as r
 from zeroconf import DNSHinfo, DNSIncoming, DNSText, const, current_time_millis
+from zeroconf._logger import _MAX_SEEN_LOGS
 from zeroconf._protocol import incoming as _incoming_module
 
 from . import has_working_ipv6
@@ -971,9 +972,9 @@ def test_seen_logs_is_bounded():
         b"\x00\x01\x00\x04\xc0\xa8\xd0\x06"
     )
     _incoming_module._seen_logs.clear()
-    for port in range(_incoming_module._MAX_SEEN_LOGS + 5):
+    for port in range(_MAX_SEEN_LOGS + 5):
         r.DNSIncoming(packet, ("1.2.3.4", port))
-    assert len(_incoming_module._seen_logs) <= _incoming_module._MAX_SEEN_LOGS
+    assert len(_incoming_module._seen_logs) <= _MAX_SEEN_LOGS
 
 
 def test_label_length_attack():
