diff --git a/tests/__init__.py b/tests/__init__.py index 86d7e1994..0e7aa930b 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -65,3 +65,4 @@ def has_working_ipv6(): def _clear_cache(zc): zc.cache.cache.clear() + zc.question_history._history.clear() diff --git a/tests/services/test_browser.py b/tests/services/test_browser.py index df405ac22..69c23cb0b 100644 --- a/tests/services/test_browser.py +++ b/tests/services/test_browser.py @@ -20,6 +20,7 @@ from zeroconf._services import ServiceStateChange from zeroconf._services.browser import ServiceBrowser from zeroconf._services.info import ServiceInfo +from zeroconf.aio import AsyncZeroconf from .. import has_working_ipv6, _inject_response @@ -426,7 +427,8 @@ def _mock_get_expiration_time(self, percent): zeroconf.close() -def test_backoff(): +@unittest.mock.patch("zeroconf._core.QuestionHistory.suppresses", return_value=False) +def test_backoff(suppresses_mock): got_query = Event() type_ = "_http._tcp.local." @@ -902,3 +904,45 @@ def test_group_ptr_queries_with_known_answers(): # If we generate multiple packets there must # only be one question assert len(packets) == 1 or len(out.questions) == 1 + + +# This test uses asyncio because it needs to access the cache directly +# which is not threadsafe +@pytest.mark.asyncio +async def test_generate_service_query_suppress_duplicate_questions(): + """Generate a service query for sending with zeroconf.send.""" + aiozc = AsyncZeroconf(interfaces=['127.0.0.1']) + zc = aiozc.zeroconf + now = current_time_millis() + name = "_hap._tcp.local." + question = r.DNSQuestion(name, const._TYPE_PTR, const._CLASS_IN) + answer = r.DNSPointer( + name, + const._TYPE_PTR, + const._CLASS_IN, + 10000, + f'known-to-other.{name}', + ) + other_known_answers = set([answer]) + zc.question_history.add_question_at_time(question, now, other_known_answers) + assert zc.question_history.suppresses(question, now, other_known_answers) + + # The known answer list is different, do not suppress + outs = _services_browser.generate_service_query(zc, now, [name], multicast=True) + assert outs + + zc.cache.async_add_records([answer]) + # The known answer list contains all the asked questions in the history + # we should suppress + + outs = _services_browser.generate_service_query(zc, now, [name], multicast=True) + assert not outs + + # We do not suppress once the question history expires + outs = _services_browser.generate_service_query(zc, now + 1000, [name], multicast=True) + assert outs + + # We do not suppress QU queries ever + outs = _services_browser.generate_service_query(zc, now, [name], multicast=False) + assert outs + await aiozc.async_close() diff --git a/tests/test_core.py b/tests/test_core.py index 87979884a..6447bbf3c 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -50,10 +50,26 @@ async def test_reaper(): record_with_10s_ttl = r.DNSAddress('a', const._TYPE_SOA, const._CLASS_IN, 10, b'a') record_with_1s_ttl = r.DNSAddress('a', const._TYPE_SOA, const._CLASS_IN, 1, b'b') zeroconf.cache.async_add_records([record_with_10s_ttl, record_with_1s_ttl]) + question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN) + now = r.current_time_millis() + other_known_answers = set( + [ + r.DNSPointer( + "_hap._tcp.local.", + const._TYPE_PTR, + const._CLASS_IN, + 10000, + 'known-to-other._hap._tcp.local.', + ) + ] + ) + zeroconf.question_history.add_question_at_time(question, now, other_known_answers) + assert zeroconf.question_history.suppresses(question, now, other_known_answers) entries_with_cache = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()])) await asyncio.sleep(1.2) entries = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()])) await aiozc.async_close() + assert not zeroconf.question_history.suppresses(question, now, other_known_answers) assert entries != original_entries assert entries_with_cache != original_entries assert record_with_10s_ttl in entries diff --git a/tests/test_handlers.py b/tests/test_handlers.py index 36f252969..d77db3f03 100644 --- a/tests/test_handlers.py +++ b/tests/test_handlers.py @@ -960,3 +960,53 @@ def async_update_records(self, zc: 'Zeroconf', now: float, records: List[r.DNSRe assert set(updated) == set([ptr_record, a_record]) await aiozc.async_close() + + +def test_questions_query_handler_populates_the_question_history_from_qm_questions(): + zc = Zeroconf(interfaces=['127.0.0.1']) + now = current_time_millis() + _clear_cache(zc) + + generated = r.DNSOutgoing(const._FLAGS_QR_QUERY) + question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN) + question.unicast = False + known_answer = r.DNSPointer( + "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.' + ) + generated.add_question(question) + generated.add_answer_at_time(known_answer, 0) + now = r.current_time_millis() + packets = generated.packets() + unicast_out, multicast_out = zc.query_handler.async_response( + [r.DNSIncoming(packet) for packet in packets], "1.2.3.4", const._MDNS_PORT + ) + assert unicast_out is None + assert multicast_out is None + assert zc.question_history.suppresses(question, now, set([known_answer])) + + zc.close() + + +def test_questions_query_handler_does_not_put_qu_questions_in_history(): + zc = Zeroconf(interfaces=['127.0.0.1']) + now = current_time_millis() + _clear_cache(zc) + + generated = r.DNSOutgoing(const._FLAGS_QR_QUERY) + question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN) + question.unicast = True + known_answer = r.DNSPointer( + "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.' + ) + generated.add_question(question) + generated.add_answer_at_time(known_answer, 0) + now = r.current_time_millis() + packets = generated.packets() + unicast_out, multicast_out = zc.query_handler.async_response( + [r.DNSIncoming(packet) for packet in packets], "1.2.3.4", const._MDNS_PORT + ) + assert unicast_out is None + assert multicast_out is None + assert not zc.question_history.suppresses(question, now, set([known_answer])) + + zc.close() diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 000000000..89159dff6 --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +"""Unit tests for _history.py.""" + +from zeroconf._history import QuestionHistory +import zeroconf as r +import zeroconf.const as const + + +def test_question_suppression(): + history = QuestionHistory() + + question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN) + now = r.current_time_millis() + other_known_answers = set( + [ + r.DNSPointer( + "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.' + ) + ] + ) + our_known_answers = set( + [ + r.DNSPointer( + "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-us._hap._tcp.local.' + ) + ] + ) + + history.add_question_at_time(question, now, other_known_answers) + + # Verify the question is suppressed if the known answers are the same + assert history.suppresses(question, now, other_known_answers) + + # Verify the question is suppressed if we know the answer to all the known answers + assert history.suppresses(question, now, other_known_answers | our_known_answers) + + # Verify the question is not suppressed if our known answers do no include the ones in the last question + assert not history.suppresses(question, now, set()) + + # Verify the question is not suppressed if our known answers do no include the ones in the last question + assert not history.suppresses(question, now, our_known_answers) + + # Verify the question is no longer suppressed after 1s + assert not history.suppresses(question, now + 1000, other_known_answers) + + +def test_question_expire(): + history = QuestionHistory() + + question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN) + now = r.current_time_millis() + other_known_answers = set( + [ + r.DNSPointer( + "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.' + ) + ] + ) + history.add_question_at_time(question, now, other_known_answers) + + # Verify the question is suppressed if the known answers are the same + assert history.suppresses(question, now, other_known_answers) + + history.async_expire(now) + + # Verify the question is suppressed if the known answers are the same since the cache hasn't expired + assert history.suppresses(question, now, other_known_answers) + + history.async_expire(now + 1000) + + # Verify the question not longer suppressed since the cache has expired + assert not history.suppresses(question, now, other_known_answers) diff --git a/zeroconf/_core.py b/zeroconf/_core.py index 1053103c0..83395f05e 100644 --- a/zeroconf/_core.py +++ b/zeroconf/_core.py @@ -35,6 +35,7 @@ from ._dns import DNSQuestion from ._exceptions import NonUniqueNameException from ._handlers import QueryHandler, RecordManager +from ._history import QuestionHistory from ._logger import QuietLogger, log from ._protocol import DNSIncoming, DNSOutgoing from ._services import RecordUpdateListener, ServiceListener @@ -134,6 +135,7 @@ async def _async_cache_cleanup(self) -> None: """Periodic cache cleanup.""" while not self.zc.done: now = current_time_millis() + self.zc.question_history.async_expire(now) self.zc.record_manager.async_updates(now, self.zc.cache.async_expire(now)) self.zc.record_manager.async_updates_complete() await asyncio.sleep(millis_to_seconds(_CACHE_CLEANUP_INTERVAL)) @@ -288,7 +290,8 @@ def __init__( self.browsers: Dict[ServiceListener, ServiceBrowser] = {} self.registry = ServiceRegistry() self.cache = DNSCache() - self.query_handler = QueryHandler(self.registry, self.cache) + self.question_history = QuestionHistory() + self.query_handler = QueryHandler(self.registry, self.cache, self.question_history) self.record_manager = RecordManager(self) self.notify_event: Optional[asyncio.Event] = None diff --git a/zeroconf/_dns.py b/zeroconf/_dns.py index e656bc519..0fad150dd 100644 --- a/zeroconf/_dns.py +++ b/zeroconf/_dns.py @@ -414,18 +414,19 @@ def __init__(self, records: Iterable[DNSRecord]) -> None: self._records = records self._lookup: Optional[Dict[DNSRecord, DNSRecord]] = None - def suppresses(self, record: DNSRecord) -> bool: - """Returns true if any answer in the rrset can suffice for the - information held in this record.""" + @property + def lookup(self) -> Dict[DNSRecord, DNSRecord]: if self._lookup is None: # Build the hash table so we can lookup the record independent of the ttl self._lookup = {record: record for record in self._records} - other = self._lookup.get(record) + return self._lookup + + def suppresses(self, record: DNSRecord) -> bool: + """Returns true if any answer in the rrset can suffice for the + information held in this record.""" + other = self.lookup.get(record) return bool(other and other.ttl > (record.ttl / 2)) def __contains__(self, record: DNSRecord) -> bool: """Returns true if the rrset contains the record.""" - if self._lookup is None: - # Build the hash table so we can lookup the record independent of the ttl - self._lookup = {record: record for record in self._records} - return record in self._lookup + return record in self.lookup diff --git a/zeroconf/_handlers.py b/zeroconf/_handlers.py index 08c25e17e..79bb6f903 100644 --- a/zeroconf/_handlers.py +++ b/zeroconf/_handlers.py @@ -21,10 +21,11 @@ """ import itertools -from typing import Dict, List, Optional, Set, TYPE_CHECKING, Tuple, Union, cast +from typing import Dict, Iterable, List, Optional, Set, TYPE_CHECKING, Tuple, Union, cast from ._cache import DNSCache, _UniqueRecordsType from ._dns import DNSAddress, DNSPointer, DNSQuestion, DNSRRSet, DNSRecord +from ._history import QuestionHistory from ._logger import log from ._protocol import DNSIncoming, DNSOutgoing from ._services import RecordUpdateListener @@ -156,10 +157,11 @@ def _has_mcast_record_in_last_second(self, record: DNSRecord) -> bool: class QueryHandler: """Query the ServiceRegistry.""" - def __init__(self, registry: ServiceRegistry, cache: DNSCache) -> None: + def __init__(self, registry: ServiceRegistry, cache: DNSCache, question_history: QuestionHistory) -> None: """Init the query handler.""" self.registry = registry self.cache = cache + self.question_history = question_history def _add_service_type_enumeration_query_answers( self, answer_set: _AnswerWithAdditionalsType, known_answers: DNSRRSet, now: float @@ -253,6 +255,8 @@ def async_response( # pylint: disable=unused-argument for msg in msgs: for question in msg.questions: + if not question.unicast: + self.question_history.add_question_at_time(question, msg.now, set(known_answers.lookup)) answer_set: _AnswerWithAdditionalsType = {} self._answer_question(question, answer_set, known_answers, msg.now) if not ucast_source and question.unicast: @@ -364,7 +368,7 @@ def async_updates_from_response(self, msg: DNSIncoming) -> None: self.async_updates_complete() def _async_mark_unique_cached_records_older_than_1s_to_expire( - self, unique_types: Set[Tuple[str, int, int]], answers: List[DNSRecord], now: float + self, unique_types: Set[Tuple[str, int, int]], answers: Iterable[DNSRecord], now: float ) -> None: # rfc6762#section-10.2 para 2 # Since unique is set, all old records with that name, rrtype, diff --git a/zeroconf/_history.py b/zeroconf/_history.py new file mode 100644 index 000000000..cbb36144a --- /dev/null +++ b/zeroconf/_history.py @@ -0,0 +1,70 @@ +""" Multicast DNS Service Discovery for Python, v0.14-wmcbrine + Copyright 2003 Paul Scott-Murphy, 2014 William McBrine + + This module provides a framework for the use of DNS Service Discovery + using IP multicast. + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License as published by the Free Software Foundation; either + version 2.1 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with this library; if not, write to the Free Software + Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 + USA +""" + +from typing import Dict, Set, Tuple + +from ._dns import DNSQuestion, DNSRecord +from .const import _DUPLICATE_QUESTION_INTERVAL + +# The QuestionHistory is used to implement Duplicate Question Suppression +# https://datatracker.ietf.org/doc/html/rfc6762#section-7.3 + + +class QuestionHistory: + def __init__(self) -> None: + self._history: Dict[DNSQuestion, Tuple[float, Set[DNSRecord]]] = {} + + def add_question_at_time(self, question: DNSQuestion, now: float, known_answers: Set[DNSRecord]) -> None: + """Remember a question with known answers.""" + self._history[question] = (now, known_answers) + + def suppresses(self, question: DNSQuestion, now: float, known_answers: Set[DNSRecord]) -> bool: + """Check to see if a question should be suppressed. + + https://datatracker.ietf.org/doc/html/rfc6762#section-7.3 + When multiple queriers on the network are querying + for the same resource records, there is no need for them to all be + repeatedly asking the same question. + """ + previous_question = self._history.get(question) + # There was not previous question in the history + if not previous_question: + return False + than, previous_known_answers = previous_question + # The last question was older than 999ms + if now - than > _DUPLICATE_QUESTION_INTERVAL: + return False + # The last question has more known answers than + # we knew so we have to ask + if previous_known_answers - known_answers: + return False + return True + + def async_expire(self, now: float) -> None: + """Expire the history of old questions.""" + removes = [ + question + for question, now_known_answers in self._history.items() + if now - now_known_answers[0] > _DUPLICATE_QUESTION_INTERVAL + ] + for question in removes: + del self._history[question] diff --git a/zeroconf/_services/browser.py b/zeroconf/_services/browser.py index 0bc926861..567ef2b68 100644 --- a/zeroconf/_services/browser.py +++ b/zeroconf/_services/browser.py @@ -31,6 +31,7 @@ from .._cache import _UniqueRecordsType from .._dns import DNSAddress, DNSPointer, DNSQuestion, DNSRecord +from .._logger import log from .._protocol import DNSOutgoing from .._services import ( RecordUpdateListener, @@ -135,11 +136,16 @@ def generate_service_query( questions_with_known_answers: _QuestionWithKnownAnswers = {} for type_ in types_: question = DNSQuestion(type_, _TYPE_PTR, _CLASS_IN) - questions_with_known_answers[question] = set( + known_answers = set( cast(DNSPointer, record) for record in zc.cache.get_all_by_details(type_, _TYPE_PTR, _CLASS_IN) if not record.is_stale(now) ) + if multicast and zc.question_history.suppresses(question, now, cast(Set[DNSRecord], known_answers)): + log.debug("Asking %s was suppressed by the question history", question) + continue + questions_with_known_answers[question] = known_answers + return _group_ptr_queries_with_known_answers(now, multicast, questions_with_known_answers) diff --git a/zeroconf/const.py b/zeroconf/const.py index ba9d5309b..df1ba8be5 100644 --- a/zeroconf/const.py +++ b/zeroconf/const.py @@ -30,6 +30,7 @@ _REGISTER_TIME = 225 # ms _LISTENER_TIME = 200 # ms _BROWSER_TIME = 1000 # ms +_DUPLICATE_QUESTION_INTERVAL = _BROWSER_TIME - 1 # ms _BROWSER_BACKOFF_LIMIT = 3600 # s _CACHE_CLEANUP_INTERVAL = 10000 # ms